Author SHA1 Message Date
sneak bd700650e0 Run at most 4 tests of a package at once in the test phase (closes #224)
check / check (push) Canceled after 0s
On a busy host, TestService_Get_ReturnsByItsDeadline failed because its
request returned over a second after its 200 ms deadline. The request
does not ignore its deadline: with as many tests running at once in the
package's test process as the host has CPUs, it waited to be scheduled,
or for SQLite calls queued on a lock inside the SQLite library that
every database in the process shares. The test phase now passes
-parallel 4 to go test. -p is unchanged: capping it also caps
compiling, which made the step much slower.

Model: opus-5-5
2026-10-08 12:20:17 +00:00
45 changed files with 248 additions and 2557 deletions
+4 -3
View File
@@ -33,9 +33,10 @@ COPY . .
# Without -v first; on a failure, again with -v for the details, and # Without -v first; on a failure, again with -v for the details, and
# the step fails even if the second run passes. -parallel 4: by default # the step fails even if the second run passes. -parallel 4: by default
# a package runs as many of its tests at once as the host has CPUs, and # a package runs as many of its tests at once as the host has CPUs. On
# on a busy host they then wait so long to be scheduled that a test # a busy host they then wait to be scheduled, and their SQLite calls
# that times a request can see it return a second late. # queue on one lock that every database in the test process shares, so
# a test that times a request can see it return a second late.
RUN go test -count=1 -timeout 90s -race -parallel 4 -cover ./... || \ RUN go test -count=1 -timeout 90s -race -parallel 4 -cover ./... || \
{ echo "--- Rerunning with -v for details ---"; \ { echo "--- Rerunning with -v for details ---"; \
go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; } go test -count=1 -timeout 90s -race -parallel 4 -v ./...; exit 1; }
+23 -69
View File
@@ -26,11 +26,6 @@ make docker
docker run -p 8080:8080 -e PIXA_SIGNING_KEY="$(openssl rand -base64 32)" pixa:latest docker run -p 8080:8080 -e PIXA_SIGNING_KEY="$(openssl rand -base64 32)" pixa:latest
``` ```
The example config allowlists no host (`allowlist_hosts`), so pixa serves an
image only for a signed or encrypted URL. Every image on an allowlisted host is
served to anyone without a signature, at any size and in any format, so
allowlist only hosts you control, where nobody else can put images.
A container takes its settings from environment variables (see Configuration A container takes its settings from environment variables (see Configuration
below for the list). Only `PIXA_SIGNING_KEY` is required; if it is unset the below for the list). Only `PIXA_SIGNING_KEY` is required; if it is unset the
container exits at startup naming the variable. Everything else has a built-in container exits at startup naming the variable. Everything else has a built-in
@@ -51,9 +46,9 @@ everything in this list without further settings. The reverse proxy must:
builds encrypted URLs from `Host`, and checks `Referer` against builds encrypted URLs from `Host`, and checks `Referer` against
`referer_blocklist`; `referer_blocklist`;
- set `X-Forwarded-For` to the client's address, with `trusted_proxies` set to - set `X-Forwarded-For` to the client's address, with `trusted_proxies` set to
the address pixa sees the proxy's requests come from, so the login limit and the address pixa sees the proxy's requests come from, so the login limit
`image_requests_per_minute` count each client by its own address (see counts each client by its own address (see `trusted_proxies` under
`trusted_proxies` under Configuration); Configuration);
- wait for pixa's answer for at least `downstream_timeout` (default `60s`), the - wait for pixa's answer for at least `downstream_timeout` (default `60s`), the
longest pixa takes to fetch, convert and send an image. longest pixa takes to fetch, convert and send an image.
@@ -63,9 +58,8 @@ password there.
pixa does the rest itself: it checks signatures and encrypted URLs, applies the pixa does the rest itself: it checks signatures and encrypted URLs, applies the
allowlist, refuses upstream hosts with private or local addresses, limits login allowlist, refuses upstream hosts with private or local addresses, limits login
attempts, each client's image requests, upstream response size and image attempts, upstream response size and image dimensions, and sends the security
dimensions, and sends the security headers, `Strict-Transport-Security` headers, `Strict-Transport-Security` included, with every response.
included, with every response.
The state directory (`state_dir`, `/var/lib/pixa` in the container) holds the The state directory (`state_dir`, `/var/lib/pixa` in the container) holds the
database and the disk cache: database and the disk cache:
@@ -89,12 +83,10 @@ which answers 200 whenever pixa is running, in maintenance mode too (see
`maintenance_mode`). `maintenance_mode`).
On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in
progress and the images being processed 5 seconds to finish, then writes to the progress and the images being processed 5 seconds to finish, and exits: with 0,
database the counts of cache hits, misses, fetches and conversions that requests or with 1 when images were still being processed after those 5 seconds or
could not write by their deadline, and exits: with 0, or with 1 when images were another part of pixa failed to stop. A request not finished by then is cut off.
still being processed after those 5 seconds, some of those counts could not be `docker stop` waits 10 seconds before it kills the container.
written, or another part of pixa failed to stop. A request not finished by then
is cut off. `docker stop` waits 10 seconds before it kills the container.
Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as
it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL
@@ -120,9 +112,7 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
- `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs and - `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs and
login, 32+ characters, for example from `openssl rand -base64 32` login, 32+ characters, for example from `openssl rand -base64 32`
- `PIXA_ALLOWLIST_HOSTS`: upstream hosts served without a signature, - `PIXA_ALLOWLIST_HOSTS`: upstream hosts served without a signature,
comma-separated; default none. Every image on a listed host is served to comma-separated
anyone without a signature, so list only hosts you control, where nobody
else can put images.
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it; - `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
default 75% of (free space + what the cache holds) default 75% of (free space + what the cache holds)
- the rest are in the table under Configuration below - the rest are in the table under Configuration below
@@ -160,14 +150,6 @@ name.
Multiple source paths may reference the same content blob; the database tracks Multiple source paths may reference the same content blob; the database tracks
references rather than using filesystem refcounting. references rather than using filesystem refcounting.
Source and transformed image files, and the `.meta` file beside each transformed
image, are written under a temporary name, synced to disk, and only then renamed
into place, so a crash leaves no file under its own name with part of its data.
When pixa starts, and every five minutes after, it removes each cached source or
transformed image whose file is empty, or not the size the database records for
it, with a warning naming the file; the next request for it fetches the source
or converts the image again.
pixa's target is 1-5k r/s, which has not been measured at that rate (see Load pixa's target is 1-5k r/s, which has not been measured at that rate (see Load
Test). Toward it, pixa keeps in memory the content types of the 10,000 Test). Toward it, pixa keeps in memory the content types of the 10,000
transformed images most recently cached or served, so a cache hit on one of them transformed images most recently cached or served, so a cache hit on one of them
@@ -204,10 +186,7 @@ path under `/v1/` answers 200, in maintenance mode too.
`Referer` names a host in `referer_blocklist`, checked before the signature, `Referer` names a host in `referer_blocklist`, checked before the signature,
the cache and the upstream fetch; 403 when the upstream host, or a host it the cache and the upstream fetch; 403 when the upstream host, or a host it
redirects to, is `localhost`, ends in `.localhost` or `.local`, or has an redirects to, is `localhost`, ends in `.localhost` or `.local`, or has an
address in a blocked network (see `blocked_networks`); 429 over address in a blocked network (see `blocked_networks`); 502 when the upstream
`image_requests_per_minute` (below); 415 when the upstream answers with a
`Content-Type` that is not one of the source formats (below), or with an image
whose first bytes libvips reads as another format; 502 when the upstream
answered with an error status, and for 5 minutes after that for the same answered with an error status, and for 5 minutes after that for the same
source URL; 503 when pixa is busy or in maintenance mode; 500 for any other source URL; 503 when pixa is busy or in maintenance mode; 500 for any other
failure. failure.
@@ -218,7 +197,7 @@ path under `/v1/` answers 200, in maintenance mode too.
expired; 504 when the upstream has not sent its response headers within expired; 504 when the upstream has not sent its response headers within
`upstream_fetch_timeout`, but 500 when that time runs out while the image `upstream_fetch_timeout`, but 500 when that time runs out while the image
itself is still arriving; 400 for an `Accept` header that is not valid, and itself is still arriving; 400 for an `Accept` header that is not valid, and
406, 403, 415, 429, 502, 503 and 500, as for `/v1/image/`. 406, 403, 502, 503 and 500, as for `/v1/image/`.
- `GET /robots.txt` — asks every crawler to stay away (`Disallow: /`). Needs: - `GET /robots.txt` — asks every crawler to stay away (`Disallow: /`). Needs:
nothing. Answers: 200. nothing. Answers: 200.
- `GET /.well-known/healthcheck.json` — JSON with `status` (`ok`), `now`, - `GET /.well-known/healthcheck.json` — JSON with `status` (`ok`), `now`,
@@ -278,17 +257,9 @@ once, is refused with 400.
both optional (values under Signature Specification). Both are part of what is both optional (values under Signature Specification). Both are part of what is
cached, so each value of either is a separate cached image. cached, so each value of either is a separate cached image.
A size with one side `0` has that side scaled to keep the source's proportions, The source image may be JPEG, PNG, GIF, WebP, AVIF, JPEG XL or SVG. The
and is refused with 400 when that side would come out above 8192, the most upstream's `Content-Type` must name its format, and the image's first bytes must
either side may be; so is a size with `fit=outside`, which scales the image to match it.
cover the size and so past it on one side, when that side would come out
above 8192.
The source image may be JPEG, PNG, GIF, WebP, AVIF or JPEG XL. The upstream's
`Content-Type` must name its format, and the image's first bytes must match it.
An upstream answering with any other `Content-Type`, SVG included, is refused
with 415, and so is an image whose first bytes libvips reads as another format,
such as an SVG whose first bytes pass the check for AVIF.
With the format `auto`, pixa chooses the format for each request from its With the format `auto`, pixa chooses the format for each request from its
`Accept` header, in this order: `Accept` header, in this order:
@@ -342,15 +313,6 @@ 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 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. address pixa sees for requests that come through the proxy closes this.
One client may send `image_requests_per_minute` requests per minute
(default 600) to the image routes, `/v1/image/` and `/v1/e/` together, `GET` and
`HEAD`, cache hits included; a request over the limit is refused with 429 and a
`Retry-After` header. Clients are counted as for the login limit. A browser's
CORS preflight is not counted. Behind a CDN or proxy whose addresses are not in
`trusted_proxies`, all requests arriving from one proxy or CDN address count as
one client, so such a deployment needs a higher limit, or `0`, which turns the
limit off.
### Encrypted URLs ### Encrypted URLs
An encrypted URL is an image URL made on pixa's own web page by someone who An encrypted URL is an image URL made on pixa's own web page by someone who
@@ -366,8 +328,7 @@ nor change what it asks for.
(`POST /generate`). The format is JPEG XL unless another is chosen; a form (`POST /generate`). The format is JPEG XL unless another is chosen; a form
sent with an empty format, or none, also makes a JPEG XL URL. Width and sent with an empty format, or none, also makes a JPEG XL URL. Width and
height both empty or `0` keep the original size; if only one of them is empty height both empty or `0` keep the original size; if only one of them is empty
or `0`, that side is scaled to keep the image's proportions. A URL whose or `0`, that side is scaled to keep the image's proportions.
scaled side would come out above 8192 answers 400.
3. The page shows the URL, `https://<host>/v1/e/<token>/img.<format>`, and when 3. The page shows the URL, `https://<host>/v1/e/<token>/img.<format>`, and when
it expires. `<host>` is the host the page was opened on, and the URL starts it expires. `<host>` is the host the page was opened on, and the URL starts
with `http` instead while `debug` is on. The name after the token is ignored with `http` instead while `debug` is on. The name after the token is ignored
@@ -521,7 +482,6 @@ tell. With no file, pixa uses the environment and the defaults.
| `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_IMAGE_REQUESTS_PER_MINUTE` | `image_requests_per_minute` | Image requests per client per minute; `0` turns it off; default `600` |
| `PIXA_UPSTREAM_FETCH_TIMEOUT` | `upstream_fetch_timeout` | Time allowed for one fetch from an upstream host; default `30s` | | `PIXA_UPSTREAM_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_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_DOWNSTREAM_TIMEOUT` | `downstream_timeout` | Time allowed for answering one client request; default `60s` |
@@ -571,13 +531,13 @@ Key settings in more detail:
explicit list replaces the default. An invalid CIDR aborts startup. Set this explicit list replaces the default. An invalid CIDR aborts startup. Set this
to the address pixa sees for requests that come through your proxy, such as to the address pixa sees for requests that come through your proxy, such as
`172.17.0.1/32`, when the defaults do not cover it, or to trust nothing else `172.17.0.1/32`, when the defaults do not cover it, or to trust nothing else
(see the login limit and `image_requests_per_minute` under Routes). For a (see the login limit under Routes). For a proxy on the Docker host that
proxy on the Docker host that connects to pixa over `127.0.0.1`, that address connects to pixa over `127.0.0.1`, that address is the gateway of the
is the gateway of the container's Docker network (`172.17.0.1` on the default container's Docker network (`172.17.0.1` on the default bridge), not the
bridge), not the proxy's own address; a proxy that connects through another of proxy's own address; a proxy that connects through another of the host's
the host's addresses is seen with that address. To be sure which address it addresses is seen with that address. To be sure which address it is, set this
is, set this to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request through the
through the proxy, and read `remoteIP` in pixa's request log line for it proxy, and read `remoteIP` in pixa's request log line for it
- `upstream_fetch_timeout` — time allowed for one fetch from an upstream host, - `upstream_fetch_timeout` — time allowed for one fetch from an upstream host,
as a duration such as `30s` (the default) or `2m` as a duration such as `30s` (the default) or `2m`
- `upstream_max_response_size` — largest upstream response accepted, in bytes; - `upstream_max_response_size` — largest upstream response accepted, in bytes;
@@ -610,12 +570,6 @@ Key settings in more detail:
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, and `downstream_timeout` has not
ended first, it is answered 503 the same way ended first, it is answered 503 the same way
- `image_requests_per_minute` — how many requests one client may send to the
image routes per minute (see the limit on them under Routes); default `600`.
`0` turns the limit off, and a negative or non-integer value aborts startup.
Behind a CDN or proxy whose addresses are not in `trusted_proxies`, all
requests arriving from one proxy or CDN address count as one client, so raise
the limit there, or set `0`
- `maintenance_mode` — while `true`, the image routes (`/v1/image/` and - `maintenance_mode` — while `true`, the image routes (`/v1/image/` and
`/v1/e/`) answer every request for an image with 503, a `Retry-After` header `/v1/e/`) answer every request for an image with 503, a `Retry-After` header
and a JSON error body. The health check (`/.well-known/healthcheck.json`) and a JSON error body. The health check (`/.well-known/healthcheck.json`)
+11 -79
View File
@@ -26,88 +26,18 @@ resolved. The disk cache is now size-bounded with LRU eviction
# Next Step # Next Step
P2: security: per-origin rate limiting P2: security: per-IP rate limiting on the image routes
# Completed Steps # Completed Steps
- 2026-10-15 a cached file left empty or partial is no longer served, or fetched - 2026-10-08 the test phase of the `Dockerfile` runs at most 4 tests of a
again on every request (closes #252). Source and transformed image files, and package at once (`go test -parallel 4`, closes #224), where it ran as many as
each transformed image's `.meta` file, are synced to disk before they are the host has CPUs. On a busy host, `TestService_Get_ReturnsByItsDeadline`
renamed into place, and their directory after; the `.meta` file is written failed because its request returned over a second after its deadline: the
before the image, and a failed directory sync removes the renamed file. The request was waiting to be scheduled, or for SQLite calls queued on a lock
reconciliation pass, at startup and every five minutes, removes a cached inside the SQLite library that every database in the test process shares.
source or transformed image whose file is empty, or not the size the database `-p`, how many packages are tested at once, is unchanged, as capping it also
records for it, with its rows and a warning naming the file, so the next slows compiling.
request fetches the source or converts the image again; requests compare no
sizes. Storing a source over a file of another size replaces that file. A
source that fails to decode is not removed for that alone: a damaged upstream
image would then be fetched on every request.
- 2026-10-15 the example config allowlists no host (closes #249): it allowlisted
`github.com` and `user-images.githubusercontent.com`, where anyone can put an
image, so a deployment set up as Getting Started says served those images to
anyone without a signature, at any size and in any format.
`configs/config.example.yml` now leaves `allowlist_hosts` out, with one
commented entry under `example.com`. Getting Started and "Running under upaas"
in `README.md` say that every image on an allowlisted host is served without a
signature, so an allowlisted host must be one the operator controls.
- 2026-10-13 `fit=outside` no longer makes an output above 8192 on a side
(closes #257): it scales the image to cover the requested size, and the side
scaled past that size was held to no maximum, so `1x8192?fit=outside` of a
3000x600 source was resized to 40960x8192 and encoded. The image processor now
refuses such a size with `ErrOutputTooLarge` before resizing, and both image
routes answer it 400, as for a size with one side `0`. The other fit modes
make the requested size or one within it, so none can go above 8192.
- 2026-10-13 a size with one side `0` no longer makes an output above 8192 on a
side (closes #241): that side is computed from the source's proportions, and
nothing held it to the 8192 a requested side is held to, so `0x8192` of a
3000x600 source was resized to 40960x8192 and encoded. The image processor now
refuses such a size with `ErrOutputTooLarge` once it knows the source's
dimensions, before resizing, and both image routes answer it 400 with the JSON
error body. The refusal is not cached as a failed fetch, so the same source is
still served at other sizes. `imgcache.MaxDimension` is now the image
processor's `MaxOutputDimension`, so the two cannot differ.
- 2026-10-13 the unused cache lifetime is removed (closes #69):
`CacheConfig.CacheTTL`, which pixad set to `DefaultCacheTTL` (24 hours), was
never read, so cached images never expired. The field and the constant are
removed. Cached source and transformed images stay until eviction removes them
to keep the cache within `cache_max_bytes`.
- 2026-10-13 SVG sources are refused, and `orig` is never saved as JPEG in place
of the source's own format (closes #68): the upstream fetch no longer accepts
`image/svg+xml`, which libvips would render. Both image routes answer 415 for
an upstream `Content-Type` that is not one of the source formats, SVG
included, where they answered 500. Before decoding, the image processor
refuses with `ErrUnsupportedInputFormat`, also answered 415, any input that
govips' `DetermineImageType`, which picks the libvips loader from the first
bytes, does not name as a source format. That catches an SVG whose first bytes
pass the first-bytes check for another type, as an XML comment can make them
do for AVIF; an SVG whose first bytes do not pass that check answers 500, as
before, and so does a damaged image in a source format, whose first bytes
`DetermineImageType` names. `orig` is always the source's own format, where a
source the processor could not name was saved as JPEG. `README.md` no longer
lists SVG. The SVG detection in `internal/magic` stays: the request path does
not call it, and it goes with the unused code of
https://git.eeqj.de/sneak/pixa/issues/168.
- 2026-10-12 per-client limit on the image routes (closes #236): the new setting
`image_requests_per_minute` (`PIXA_IMAGE_REQUESTS_PER_MINUTE`), default 600,
is how many `GET` and `HEAD` requests one client may send to `/v1/image/` and
`/v1/e/` together per minute, cache hits included; the next is answered 429
with `Retry-After`. It uses the login limit's `RateLimit` middleware, so
clients are told apart the same way, and runs after the CORS middleware, so a
preflight is not counted. `0` turns the limit off; a negative or non-integer
value aborts startup.
- 2026-10-08 a request past its deadline no longer waits for the database to
count it (closes #224). On a busy host, `TestService_Get_ReturnsByItsDeadline`
failed because its request, past its deadline, still waited for its miss count
to be written, a write with no deadline, which in pixad waits for the one
database connection every request shares. Each count write now keeps the
request's deadline but not its cancellation. A count not written by then is
kept in memory, where `Stats` sees it, and one goroutine of the cache writes
those counts, one write at a time, and once more at shutdown before the
database closes. The test phase of the `Dockerfile` also runs at most 4 tests
of a package at once (`go test -parallel 4`), where it ran as many as the host
has CPUs: on a busy host they then wait so long to be scheduled that a test
that times a request can fail. `-p`, how many packages are tested at once, is
unchanged, as capping it also slows compiling.
- 2026-10-08 every output format is saved with settings pixa sets on purpose - 2026-10-08 every output format is saved with settings pixa sets on purpose
(closes #232): each format has its own govips export, as JPEG XL does, in (closes #232): each format has its own govips export, as JPEG XL does, in
place of govips' generic `Export`, which sent libvips a zero for some settings place of govips' generic `Export`, which sent libvips a zero for some settings
@@ -814,6 +744,8 @@ P2: security: per-origin rate limiting
# Future Steps # Future Steps
- P2: security
- per-origin rate limiting
- P2: HTTP response handling - P2: HTTP response handling
- Last-Modified headers - Last-Modified headers
- P2: configuration - P2: configuration
+6 -13
View File
@@ -45,14 +45,15 @@ state_dir: ./data
signing_key: "CHANGE_ME_generate_with_openssl_rand_base64_32" signing_key: "CHANGE_ME_generate_with_openssl_rand_base64_32"
# Hosts that don't require signatures (default: none) # Hosts that don't require signatures (default: none)
# Every image on an allowlisted host is served to anyone without a
# signature, at any size and in any format, so list only hosts you
# control, where nobody else can put images.
# Use "." prefix for wildcard subdomain matching (e.g., ".example.com" matches "cdn.example.com") # Use "." prefix for wildcard subdomain matching (e.g., ".example.com" matches "cdn.example.com")
# An entry that is neither a host name nor an IP address (IPv6 without # An entry that is neither a host name nor an IP address (IPv6 without
# brackets), such as one with a port or a "*." wildcard, aborts startup. # brackets), such as one with a port or a "*." wildcard, aborts startup.
# allowlist_hosts: allowlist_hosts:
# - images.example.com - s3.sneak.cloud
- static.sneak.cloud
- sneak.berlin
- github.com
- user-images.githubusercontent.com
# Hosts whose pages may not show pixa's images, written as for # Hosts whose pages may not show pixa's images, written as for
# allowlist_hosts. A request to /v1/image/ or /v1/e/ whose Referer header # allowlist_hosts. A request to /v1/image/ or /v1/e/ whose Referer header
@@ -114,14 +115,6 @@ upstream_connections: 64
# frees up it is answered 503, unless downstream_timeout has ended first. # frees up it is answered 503, unless downstream_timeout has ended first.
# max_concurrent_processing: 4 # max_concurrent_processing: 4
# How many requests to the image routes (/v1/image/ and /v1/e/) one client
# may send per minute; a request over the limit is answered 429 with a
# Retry-After header. 0 turns the limit off. Clients are told apart as for
# the login limit, so behind a CDN or proxy whose addresses are not in
# trusted_proxies all requests arriving from one proxy or CDN address count
# as one client: raise the limit there, or set 0. (default: 600)
image_requests_per_minute: 600
# Time allowed for one fetch from an upstream host (default: 30s) # Time allowed for one fetch from an upstream host (default: 30s)
upstream_fetch_timeout: 30s upstream_fetch_timeout: 30s
+10 -39
View File
@@ -34,7 +34,6 @@ const (
DefaultUpstreamFetchTimeout = 30 * time.Second DefaultUpstreamFetchTimeout = 30 * time.Second
DefaultUpstreamMaxResponseSize = 50 << 20 // 50 MiB DefaultUpstreamMaxResponseSize = 50 << 20 // 50 MiB
DefaultDownstreamTimeout = 60 * time.Second DefaultDownstreamTimeout = 60 * time.Second
DefaultImageRequestsPerMinute = 600
) )
// Configuration key names. // Configuration key names.
@@ -62,7 +61,6 @@ const (
keyUpstreamFetchTimeout = "upstream_fetch_timeout" keyUpstreamFetchTimeout = "upstream_fetch_timeout"
keyUpstreamMaxResponseSize = "upstream_max_response_size" keyUpstreamMaxResponseSize = "upstream_max_response_size"
keyDownstreamTimeout = "downstream_timeout" keyDownstreamTimeout = "downstream_timeout"
keyImageRequestsPerMinute = "image_requests_per_minute"
) )
// placeholderSigningKey is the dummy signing_key shipped in // placeholderSigningKey is the dummy signing_key shipped in
@@ -160,11 +158,6 @@ type Config struct {
// of the per-request timeout middleware. // of the per-request timeout middleware.
DownstreamTimeout time.Duration DownstreamTimeout time.Duration
// ImageRequestsPerMinute is how many requests one client may make to
// the image routes per minute; the next is refused with 429 Too Many
// Requests. 0 turns the limit off.
ImageRequestsPerMinute int
// 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.
@@ -309,8 +302,6 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
keyAccessControlAllowOrigin, DefaultAccessControlAllowOrigin), keyAccessControlAllowOrigin, DefaultAccessControlAllowOrigin),
DownstreamTimeout: loader.durationVal( DownstreamTimeout: loader.durationVal(
keyDownstreamTimeout, DefaultDownstreamTimeout), keyDownstreamTimeout, DefaultDownstreamTimeout),
ImageRequestsPerMinute: loader.intVal(
keyImageRequestsPerMinute, DefaultImageRequestsPerMinute),
CacheMaxBytes: loader.int64Val(keyCacheMaxBytes, 0), CacheMaxBytes: loader.int64Val(keyCacheMaxBytes, 0),
BlockedNetworks: blockedNetworks, BlockedNetworks: blockedNetworks,
TrustedProxies: trustedProxies, TrustedProxies: trustedProxies,
@@ -320,7 +311,9 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
// The default for an omitted cache_max_bytes is worked out when // The default for an omitted cache_max_bytes is worked out when
// the cache opens; here we only record whether the operator set // the cache opens; here we only record whether the operator set
// the key explicitly. // the key explicitly.
_, c.CacheMaxBytesExplicit = lookupValue(sc, keyCacheMaxBytes) if _, present := lookupValue(sc, keyCacheMaxBytes); present {
c.CacheMaxBytesExplicit = true
}
// Build DBURL from StateDir if not explicitly set. The derived URL // Build DBURL from StateDir if not explicitly set. The derived URL
// is a default: it applies only when db_url is omitted, never to an // is a default: it applies only when db_url is omitted, never to an
@@ -433,7 +426,7 @@ func isKnownConfigKey(key string) bool {
keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks, keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks,
keyTrustedProxies, keyAccessControlAllowOrigin, keyUpstreamFetchTimeout, keyTrustedProxies, keyAccessControlAllowOrigin, keyUpstreamFetchTimeout,
keyUpstreamMaxResponseSize, keyDownstreamTimeout, keyRefererBlocklist, keyUpstreamMaxResponseSize, keyDownstreamTimeout, keyRefererBlocklist,
keyImageRequestsPerMinute, "env": "env":
return true return true
} }
@@ -468,7 +461,6 @@ func envVarNames() map[string]string {
keyUpstreamFetchTimeout: "PIXA_UPSTREAM_FETCH_TIMEOUT", keyUpstreamFetchTimeout: "PIXA_UPSTREAM_FETCH_TIMEOUT",
keyUpstreamMaxResponseSize: "PIXA_UPSTREAM_MAX_RESPONSE_SIZE", keyUpstreamMaxResponseSize: "PIXA_UPSTREAM_MAX_RESPONSE_SIZE",
keyDownstreamTimeout: "PIXA_DOWNSTREAM_TIMEOUT", keyDownstreamTimeout: "PIXA_DOWNSTREAM_TIMEOUT",
keyImageRequestsPerMinute: "PIXA_IMAGE_REQUESTS_PER_MINUTE",
} }
} }
@@ -621,14 +613,6 @@ func (c *Config) validate() error {
settingName(keyCacheMaxBytes), c.CacheMaxBytes, errMustNotBeNegative) settingName(keyCacheMaxBytes), c.CacheMaxBytes, errMustNotBeNegative)
} }
// Zero is valid (it turns the limit off); only negative values are
// rejected.
if c.ImageRequestsPerMinute < 0 {
return fmt.Errorf("%s: value %d %w",
settingName(keyImageRequestsPerMinute), c.ImageRequestsPerMinute,
errMustNotBeNegative)
}
err = c.validateUpstreamMaxResponseSize() err = c.validateUpstreamMaxResponseSize()
if err != nil { if err != nil {
return err return err
@@ -641,9 +625,12 @@ func (c *Config) validate() error {
} }
} }
err = c.validateSentryDSN() if c.SentryDSN != "" {
if err != nil { parsed, err := url.Parse(c.SentryDSN)
return err if err != nil || parsed.Scheme == "" || parsed.Host == "" {
return fmt.Errorf("%s: value %q is %w",
settingName(keySentryDSN), c.SentryDSN, errNotAValidURL)
}
} }
if (c.MetricsUsername == "") != (c.MetricsPassword == "") { if (c.MetricsUsername == "") != (c.MetricsPassword == "") {
@@ -655,22 +642,6 @@ func (c *Config) validate() error {
return c.validateAccessControlAllowOrigin() return c.validateAccessControlAllowOrigin()
} }
// validateSentryDSN checks that sentry_dsn, when set, is a URL with a scheme
// and a host.
func (c *Config) validateSentryDSN() error {
if c.SentryDSN == "" {
return nil
}
parsed, err := url.Parse(c.SentryDSN)
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
return fmt.Errorf("%s: value %q is %w",
settingName(keySentryDSN), c.SentryDSN, errNotAValidURL)
}
return nil
}
// validateUpstreamMaxResponseSize checks that upstream_max_response_size // validateUpstreamMaxResponseSize checks that upstream_max_response_size
// is from 1 byte to 1 GiB. An upstream response is read whole into // 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 // memory, and the image processor reads one byte past this limit, which
-26
View File
@@ -92,32 +92,6 @@ func TestGetStringSlice_Empty(t *testing.T) {
} }
} }
// TestExampleConfigAllowlistsNoHost loads configs/config.example.yml
// with a real signing key, as Getting Started in README.md has it copied
// and edited, and checks that it allowlists no host: every image on an
// allowlisted host is served to anyone without a signature.
func TestExampleConfigAllowlistsNoHost(t *testing.T) {
t.Setenv("PIXA_SIGNING_KEY", validTestSigningKey)
examplePath := filepath.Join("..", "..", "configs", "config.example.yml")
sc, err := loadTestConfig(examplePath)
if err != nil {
t.Fatalf("failed to load %s: %v", examplePath, err)
}
c, err := newFromSmartConfig(sc)
if err != nil {
t.Fatalf("%s is not a valid config: %v", examplePath, err)
}
t.Logf("allowlist_hosts in %s: %v", examplePath, c.AllowlistHosts)
if len(c.AllowlistHosts) != 0 {
t.Errorf("%s allowlists %v, want no host", examplePath, c.AllowlistHosts)
}
}
// loadTestConfig is a helper to load a config file for testing. // loadTestConfig is a helper to load a config file for testing.
func loadTestConfig(path string) (*smartconfig.Config, error) { func loadTestConfig(path string) (*smartconfig.Config, error) {
return smartconfig.NewFromConfigPath(path) return smartconfig.NewFromConfigPath(path)
-2
View File
@@ -70,7 +70,6 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
t.Setenv("PIXA_UPSTREAM_CONNECTIONS_PER_HOST", "5") t.Setenv("PIXA_UPSTREAM_CONNECTIONS_PER_HOST", "5")
t.Setenv("PIXA_UPSTREAM_CONNECTIONS", "10") t.Setenv("PIXA_UPSTREAM_CONNECTIONS", "10")
t.Setenv("PIXA_MAX_CONCURRENT_PROCESSING", "3") t.Setenv("PIXA_MAX_CONCURRENT_PROCESSING", "3")
t.Setenv("PIXA_IMAGE_REQUESTS_PER_MINUTE", "30")
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")
@@ -100,7 +99,6 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
UpstreamConnectionsPerHost: 5, UpstreamConnectionsPerHost: 5,
UpstreamConnections: 10, UpstreamConnections: 10,
MaxConcurrentProcessing: 3, MaxConcurrentProcessing: 3,
ImageRequestsPerMinute: 30,
CacheMaxBytes: 1024, CacheMaxBytes: 1024,
CacheMaxBytesExplicit: true, CacheMaxBytesExplicit: true,
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")}, BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")},
@@ -1,107 +0,0 @@
package config
import "testing"
// testImageRequestsPerMinuteVar is the variable that sets
// image_requests_per_minute.
const testImageRequestsPerMinuteVar = "PIXA_IMAGE_REQUESTS_PER_MINUTE"
// TestImageRequestsPerMinuteValues checks that an omitted
// image_requests_per_minute is 600, and that 0, which turns the limit off,
// and any larger whole number are used as given.
func TestImageRequestsPerMinuteValues(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.ImageRequestsPerMinute != 600 {
t.Errorf("ImageRequestsPerMinute = %d, want 600", c.ImageRequestsPerMinute)
}
explicit := []struct {
value string
want int
}{
{"0", 0},
{"30", 30},
}
for _, tc := range explicit {
c, err := configFromYAML(t,
signingKeyLine+"image_requests_per_minute: "+tc.value+"\n")
if err != nil {
t.Fatalf("image_requests_per_minute: %s should load, got error: %v",
tc.value, err)
}
if c.ImageRequestsPerMinute != tc.want {
t.Errorf("ImageRequestsPerMinute = %d, want %d",
c.ImageRequestsPerMinute, tc.want)
}
}
}
// TestInvalidImageRequestsPerMinuteAbortsStartup checks that a value that is
// negative, not a whole number, or null aborts startup naming the key and
// the value, and the variable too where the value could have come from it.
func TestInvalidImageRequestsPerMinuteAbortsStartup(t *testing.T) {
t.Parallel()
key := keyImageRequestsPerMinute
runAbortCases(t, []abortCase{
{
name: "negative",
yaml: signingKeyLine + key + ": -1\n",
wantErrSubstrings: []string{
key, testImageRequestsPerMinuteVar, "value -1",
},
},
{
name: "not a number",
yaml: signingKeyLine + key + ": plenty\n",
wantErrSubstrings: []string{
key, testImageRequestsPerMinuteVar, "plenty",
},
},
{
name: "fractional",
yaml: signingKeyLine + key + ": 2.5\n",
wantErrSubstrings: []string{key, "2.5"},
},
{
name: "explicit null",
yaml: signingKeyLine + key + ": null\n",
wantErrSubstrings: []string{key, nullValueText},
},
})
}
// TestImageRequestsPerMinuteFromEnvironment checks that the variable sets the
// limit over the config file, and that an invalid value in it aborts startup
// naming the variable and the value.
func TestImageRequestsPerMinuteFromEnvironment(t *testing.T) {
t.Setenv(testImageRequestsPerMinuteVar, "0")
c, err := configFromYAML(t, signingKeyLine+"image_requests_per_minute: 30\n")
if err != nil {
t.Fatalf("limit from the environment should load: %v", err)
}
if c.ImageRequestsPerMinute != 0 {
t.Errorf("ImageRequestsPerMinute = %d, want 0 from the environment",
c.ImageRequestsPerMinute)
}
for _, value := range []string{"-5", "ten"} {
t.Run(value, func(t *testing.T) {
t.Setenv(testImageRequestsPerMinuteVar, value)
_, err := configFromYAML(t, signingKeyLine)
wantStartupError(t, err, testImageRequestsPerMinuteVar, value)
})
}
}
+3 -11
View File
@@ -4,7 +4,6 @@ package handlers
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"log/slog" "log/slog"
"net/http" "net/http"
"time" "time"
@@ -72,21 +71,16 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
} }
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // the cache's goroutines outlive OnStart; OnStop stops them //nolint:contextcheck // the eviction loop outlives OnStart; OnStop cancels it
OnStart: func(_ context.Context) error { OnStart: func(_ context.Context) error {
return s.initImageService() return s.initImageService()
}, },
// The pending counts are written here, before the database's stop
// hook closes it.
OnStop: func(ctx context.Context) error { OnStop: func(ctx context.Context) error {
if s.imgCache == nil { if s.imgCache == nil {
return nil return nil
} }
return errors.Join( return s.imgCache.StopEviction(ctx)
s.imgCache.StopEviction(ctx),
s.imgCache.StopPendingCountWrites(ctx),
)
}, },
}) })
@@ -106,6 +100,7 @@ func (s *Handlers) WaitForProcessing(ctx context.Context) int {
func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig { func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig {
return imgcache.CacheConfig{ return imgcache.CacheConfig{
StateDir: cfg.StateDir, StateDir: cfg.StateDir,
CacheTTL: imgcache.DefaultCacheTTL,
NegativeTTL: imgcache.DefaultNegativeTTL, NegativeTTL: imgcache.DefaultNegativeTTL,
MaxBytes: cfg.CacheMaxBytes, MaxBytes: cfg.CacheMaxBytes,
UseDefaultMaxBytes: !cfg.CacheMaxBytesExplicit, UseDefaultMaxBytes: !cfg.CacheMaxBytesExplicit,
@@ -127,9 +122,6 @@ func (s *Handlers) initImageService() error {
// write-pressure passes. No-op when the disk cache is disabled. // write-pressure passes. No-op when the disk cache is disabled.
cache.StartEviction(imgcache.DefaultEvictionInterval) cache.StartEviction(imgcache.DefaultEvictionInterval)
// Writes the counts requests could not write by their deadline
cache.StartPendingCountWrites()
// Create the fetcher config // Create the fetcher config
fetcherCfg := httpfetcher.DefaultConfig() fetcherCfg := httpfetcher.DefaultConfig()
fetcherCfg.AllowHTTP = s.config.AllowHTTP fetcherCfg.AllowHTTP = s.config.AllowHTTP
@@ -47,6 +47,7 @@ func setupTestHandler(t *testing.T) *testFixtures {
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{ cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
-26
View File
@@ -206,16 +206,6 @@ func parseExpires(query url.Values) (time.Time, error) {
return time.Unix(exp, 0), nil return time.Unix(exp, 0), nil
} }
// HandleTooManyRequests answers 429 Too Many Requests with the JSON error
// body the image routes send for their other errors. The image routes' rate
// limit calls it for a request over the limit.
func (s *Handlers) HandleTooManyRequests() http.HandlerFunc {
return func(w http.ResponseWriter, _ *http.Request) {
s.respondError(w, "too many requests, try again later",
http.StatusTooManyRequests)
}
}
// respondImageError maps image retrieval errors to HTTP responses. // respondImageError maps image retrieval errors to HTTP responses.
func (s *Handlers) respondImageError( func (s *Handlers) respondImageError(
w http.ResponseWriter, req *imgcache.ImageRequest, err error, w http.ResponseWriter, req *imgcache.ImageRequest, err error,
@@ -247,22 +237,6 @@ func (s *Handlers) respondImageError(
return return
} }
if errors.Is(err, httpfetcher.ErrInvalidContentType) ||
errors.Is(err, imageprocessor.ErrUnsupportedInputFormat) {
s.respondError(w, "unsupported source format",
http.StatusUnsupportedMediaType)
return
}
if errors.Is(err, imageprocessor.ErrOutputTooLarge) {
s.respondError(w, fmt.Sprintf("size too large for the source: the side "+
"computed from its proportions is above %d",
imageprocessor.MaxOutputDimension), http.StatusBadRequest)
return
}
s.respondError(w, "internal error", http.StatusInternalServerError) s.respondError(w, "internal error", http.StatusInternalServerError)
} }
@@ -31,6 +31,7 @@ func newSignedHostServer(
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
@@ -29,6 +29,7 @@ func newImageRoute(t *testing.T, fetcher httpfetcher.Fetcher) http.Handler {
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
-8
View File
@@ -161,14 +161,6 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
errors.Is(err, imageprocessor.ErrTooManyImages): errors.Is(err, imageprocessor.ErrTooManyImages):
s.respondError(w, "server busy, try again later", s.respondError(w, "server busy, try again later",
http.StatusServiceUnavailable) http.StatusServiceUnavailable)
case errors.Is(err, httpfetcher.ErrInvalidContentType),
errors.Is(err, imageprocessor.ErrUnsupportedInputFormat):
s.respondError(w, "unsupported source format",
http.StatusUnsupportedMediaType)
case errors.Is(err, imageprocessor.ErrOutputTooLarge):
s.respondError(w, fmt.Sprintf("size too large for the source: the side "+
"computed from its proportions is above %d",
imageprocessor.MaxOutputDimension), http.StatusBadRequest)
default: default:
s.log.Error("image request failed", "error", err) s.log.Error("image request failed", "error", err)
s.respondError(w, "internal error", http.StatusInternalServerError) s.respondError(w, "internal error", http.StatusInternalServerError)
@@ -49,6 +49,7 @@ func newRefererRoutes(t *testing.T) (http.Handler, *Handlers, *countingFetcher)
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{ cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
@@ -1,54 +0,0 @@
package handlers
import (
"fmt"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"testing"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/imgcache"
)
// TestUnsupportedSourceFormatAnswers415 checks that both image routes answer
// 415 with a clear error when the upstream answers with a content type pixa
// does not accept, such as SVG, and when orig is asked of a source in a
// format the processor cannot name, wrapped as the service wraps them.
func TestUnsupportedSourceFormatAnswers415(t *testing.T) {
t.Parallel()
h := &Handlers{log: slog.New(slog.DiscardHandler)}
req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.svg"}
for _, err := range []error{
fmt.Errorf("upstream fetch failed: %w: image/svg+xml",
httpfetcher.ErrInvalidContentType),
fmt.Errorf("image processing failed: %w: unknown",
imageprocessor.ErrUnsupportedInputFormat),
} {
plain := httptest.NewRecorder()
h.respondImageError(plain, req, err)
encrypted := httptest.NewRecorder()
h.handleImageError(encrypted, err)
for route, rec := range map[string]*httptest.ResponseRecorder{
"HandleImage": plain, "HandleImageEnc": encrypted,
} {
t.Logf("%s for %v: %d %s", route, err, rec.Code, rec.Body.String())
if rec.Code != http.StatusUnsupportedMediaType {
t.Errorf("%s for %v: status = %d, want %d",
route, err, rec.Code, http.StatusUnsupportedMediaType)
}
if !strings.Contains(rec.Body.String(), "unsupported source format") {
t.Errorf("%s for %v: body = %q, want the unsupported source format error",
route, err, rec.Body.String())
}
}
}
}
+1
View File
@@ -158,6 +158,7 @@ func DefaultConfig() *Config {
contentTypeWebP, contentTypeWebP,
contentTypeAVIF, contentTypeAVIF,
contentTypeJXL, contentTypeJXL,
contentTypeSVG,
}, },
AllowHTTP: false, AllowHTTP: false,
MaxConnectionsPerHost: DefaultMaxConnectionsPerHost, MaxConnectionsPerHost: DefaultMaxConnectionsPerHost,
@@ -1,52 +0,0 @@
package imageprocessor
import (
"bytes"
"errors"
"testing"
)
// TestImageProcessor_RefusesComputedSideAboveMaximum verifies that a size
// with one side 0, which Process computes from the input's proportions, is
// served when that side comes out at MaxOutputDimension and refused with
// ErrOutputTooLarge when it would be above. A 4096x1 or 1x4096 input given
// 2 for its short side has 8192 for its long side.
func TestImageProcessor_RefusesComputedSideAboveMaximum(t *testing.T) {
t.Parallel()
tests := []struct {
name string
inputWidth, inputHeight int
size Size
wantErr error
}{
{"width at the maximum", 4096, 1, Size{Height: 2}, nil},
{"width above the maximum", 4096, 1, Size{Height: 3}, ErrOutputTooLarge},
{"height at the maximum", 1, 4096, Size{Width: 2}, nil},
{"height above the maximum", 1, 4096, Size{Width: 3}, ErrOutputTooLarge},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(),
bytes.NewReader(input),
&Request{Size: tt.size, Format: FormatJPEG, FitMode: FitCover})
if result != nil {
t.Logf("Process() made %dx%d", result.Width, result.Height)
if max(result.Width, result.Height) != MaxOutputDimension {
t.Errorf("Process() made %dx%d, want a side of %d",
result.Width, result.Height, MaxOutputDimension)
}
}
if !errors.Is(err, tt.wantErr) {
t.Errorf("Process() error = %v, want %v", err, tt.wantErr)
}
})
}
}
@@ -1,62 +0,0 @@
package imageprocessor
import (
"bytes"
"errors"
"testing"
)
// TestImageProcessor_RefusesFitOutsideAboveMaximum verifies that FitOutside,
// which scales the input to cover the requested size and so past it on one
// side, is refused with ErrOutputTooLarge when that side would come out above
// MaxOutputDimension, and otherwise makes the scaled size. A 1024x256 or
// 256x1024 input scaled by 8 has 8192 for its long side.
func TestImageProcessor_RefusesFitOutsideAboveMaximum(t *testing.T) {
t.Parallel()
tests := []struct {
name string
inputWidth, inputHeight int
size Size
// wantWidth and wantHeight are the output's, 0 for a refusal
wantWidth, wantHeight int
}{
{"width above the maximum", 3000, 600, Size{Width: 1, Height: 8192}, 0, 0},
{"height above the maximum", 600, 3000, Size{Width: 8192, Height: 1}, 0, 0},
{"width at the maximum", 1024, 256, Size{Width: 1, Height: 2048}, 8192, 2048},
{"height at the maximum", 256, 1024, Size{Width: 2048, Height: 1}, 2048, 8192},
{"within the maximum", 3000, 600, Size{Width: 1, Height: 1200}, 6000, 1200},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
input := createTestJPEG(t, tt.inputWidth, tt.inputHeight)
result, err := New(Params{}).Process(t.Context(),
bytes.NewReader(input),
&Request{Size: tt.size, Format: FormatJPEG, FitMode: FitOutside})
if result != nil {
t.Logf("Process() made %dx%d", result.Width, result.Height)
}
if tt.wantWidth == 0 {
if !errors.Is(err, ErrOutputTooLarge) {
t.Errorf("Process() error = %v, want %v", err, ErrOutputTooLarge)
}
return
}
if err != nil {
t.Fatalf("Process() error = %v", err)
}
if result.Width != tt.wantWidth || result.Height != tt.wantHeight {
t.Errorf("Process() made %dx%d, want %dx%d",
result.Width, result.Height, tt.wantWidth, tt.wantHeight)
}
})
}
}
+30 -73
View File
@@ -133,26 +133,10 @@ var ErrInputTooLarge = errors.New("input image dimensions exceed maximum")
// configured byte limit. // configured byte limit.
var ErrInputDataTooLarge = errors.New("input data exceeds maximum allowed size") var ErrInputDataTooLarge = errors.New("input data exceeds maximum allowed size")
// MaxOutputDimension is the maximum width or height of an output image, and
// so of a requested size (imgcache.MaxDimension).
const MaxOutputDimension = 8192
// ErrOutputTooLarge is returned, before resizing, when the output would have a
// side above MaxOutputDimension: the side of a size with one side 0, computed
// from the input's proportions, or with FitOutside, the side scaled past the
// requested size.
var ErrOutputTooLarge = errors.New("output image dimensions exceed maximum")
// ErrUnsupportedOutputFormat is returned when the requested output format is // ErrUnsupportedOutputFormat is returned when the requested output format is
// not supported. // not supported.
var ErrUnsupportedOutputFormat = errors.New("unsupported output format") var ErrUnsupportedOutputFormat = errors.New("unsupported output format")
// ErrUnsupportedInputFormat is returned, before the input is decoded, when
// libvips reads the input's first bytes as another format than the source
// formats, or as none: for example an SVG whose first bytes pass the
// first-bytes check for AVIF.
var ErrUnsupportedInputFormat = errors.New("unsupported input format")
// ErrTooManyImages is returned when MaxConcurrentProcessing images are being // ErrTooManyImages is returned when MaxConcurrentProcessing images are being
// processed and none finishes within ProcessingWaitTimeout. // processed and none finishes within ProcessingWaitTimeout.
var ErrTooManyImages = errors.New("too many images being processed at once") var ErrTooManyImages = errors.New("too many images being processed at once")
@@ -235,17 +219,6 @@ func (p *ImageProcessor) Process(
return nil, ErrInputDataTooLarge return nil, ErrInputDataTooLarge
} }
// govips picks the libvips loader from the input's first bytes with
// DetermineImageType. Input whose first bytes it reads as no source
// format, such as SVG, which librsvg would render, is refused before
// libvips reads it.
inputFormat := p.detectFormat(vips.DetermineImageType(data))
sourceFormat, err := p.formatFromString(inputFormat)
if err != nil {
return nil, err
}
// Decode image // Decode image
img, err := vips.NewImageFromBuffer(data) img, err := vips.NewImageFromBuffer(data)
if err != nil { if err != nil {
@@ -264,16 +237,16 @@ func (p *ImageProcessor) Process(
origWidth := img.Width() origWidth := img.Width()
origHeight := img.Height() origHeight := img.Height()
// Detect input format
inputFormat := p.detectFormat(img)
// Validate input dimensions to prevent DoS via decompression bombs // Validate input dimensions to prevent DoS via decompression bombs
if origWidth > MaxInputDimension || origHeight > MaxInputDimension { if origWidth > MaxInputDimension || origHeight > MaxInputDimension {
return nil, ErrInputTooLarge return nil, ErrInputTooLarge
} }
// Determine target dimensions; too large ones are refused before resizing // Determine target dimensions
targetWidth, targetHeight, err := targetDimensions(req.Size, origWidth, origHeight) targetWidth, targetHeight := targetDimensions(req.Size, origWidth, origHeight)
if err != nil {
return nil, err
}
// Resize if needed // Resize if needed
if targetWidth != origWidth || targetHeight != origHeight { if targetWidth != origWidth || targetHeight != origHeight {
@@ -286,7 +259,7 @@ func (p *ImageProcessor) Process(
// orig is the source's own format; encode refuses an empty format // orig is the source's own format; encode refuses an empty format
outputFormat := req.Format outputFormat := req.Format
if outputFormat == FormatOriginal { if outputFormat == FormatOriginal {
outputFormat = sourceFormat outputFormat = p.formatFromString(inputFormat)
} }
// Encode to target format // Encode to target format
@@ -309,29 +282,21 @@ func (p *ImageProcessor) Process(
// targetDimensions calculates the output dimensions for a requested size, // targetDimensions calculates the output dimensions for a requested size,
// scaling proportionally when only one dimension is given and keeping the // scaling proportionally when only one dimension is given and keeping the
// original dimensions when both are zero. A side scaled proportionally can // original dimensions when both are zero.
// come out above MaxOutputDimension, which is ErrOutputTooLarge. func targetDimensions(size Size, origWidth, origHeight int) (int, int) {
func targetDimensions(size Size, origWidth, origHeight int) (int, int, error) {
width, height := size.Width, size.Height
switch { switch {
case width == 0 && height == 0: case size.Width == 0 && size.Height == 0:
// Both are 0: keep original size // Both are 0: keep original size
return origWidth, origHeight, nil return origWidth, origHeight
case width == 0: case size.Width == 0:
// Only height specified: calculate width proportionally // Only height specified: calculate width proportionally
width = origWidth * height / origHeight return origWidth * size.Height / origHeight, size.Height
case height == 0: case size.Height == 0:
// Only width specified: calculate height proportionally // Only width specified: calculate height proportionally
height = origHeight * width / origWidth return size.Width, origHeight * size.Width / origWidth
default:
return size.Width, size.Height
} }
if width > MaxOutputDimension || height > MaxOutputDimension {
return 0, 0, fmt.Errorf("%w: %dx%d is above %d", ErrOutputTooLarge,
width, height, MaxOutputDimension)
}
return width, height, nil
} }
// MIME types for the supported image formats. // MIME types for the supported image formats.
@@ -438,9 +403,10 @@ func (p *ImageProcessor) acquireSlot(ctx context.Context) (func(), error) {
} }
} }
// detectFormat returns the format string for a govips image type. // detectFormat returns the format string from a vips image.
func (p *ImageProcessor) detectFormat(imageType vips.ImageType) string { func (p *ImageProcessor) detectFormat(img *vips.ImageRef) string {
switch imageType { format := img.Format()
switch format {
case vips.ImageTypeJPEG: case vips.ImageTypeJPEG:
return "jpeg" return "jpeg"
case vips.ImageTypePNG: case vips.ImageTypePNG:
@@ -462,9 +428,7 @@ func (p *ImageProcessor) detectFormat(imageType vips.ImageType) string {
} }
} }
// resize resizes the image according to the fit mode. With FitOutside one // resize resizes the image according to the fit mode.
// side can come out larger than requested; when it would be above
// MaxOutputDimension, resize returns ErrOutputTooLarge without resizing.
func (p *ImageProcessor) resize( func (p *ImageProcessor) resize(
img *vips.ImageRef, width, height int, fit FitMode, img *vips.ImageRef, width, height int, fit FitMode,
) error { ) error {
@@ -512,12 +476,6 @@ func (p *ImageProcessor) resize(
newW := int(float64(imgW) * scale) newW := int(float64(imgW) * scale)
newH := int(float64(imgH) * scale) newH := int(float64(imgH) * scale)
// The dimension that exceeds the target is held to the maximum too
if newW > MaxOutputDimension || newH > MaxOutputDimension {
return fmt.Errorf("%w: %dx%d is above %d", ErrOutputTooLarge,
newW, newH, MaxOutputDimension)
}
return img.Thumbnail(newW, newH, vips.InterestingNone) return img.Thumbnail(newW, newH, vips.InterestingNone)
default: default:
@@ -729,23 +687,22 @@ func jxlDistance(quality int) float64 {
return 53.0/3000.0*q*q - 23.0/20.0*q + 25.0 return 53.0/3000.0*q*q - 23.0/20.0*q + 25.0
} }
// formatFromString converts a format string from detectFormat to Format. A // formatFromString converts a format string to Format.
// format detectFormat could not name is ErrUnsupportedInputFormat. func (p *ImageProcessor) formatFromString(format string) Format {
func (p *ImageProcessor) formatFromString(format string) (Format, error) {
switch format { switch format {
case "jpeg": case "jpeg":
return FormatJPEG, nil return FormatJPEG
case "png": case "png":
return FormatPNG, nil return FormatPNG
case "gif": case "gif":
return FormatGIF, nil return FormatGIF
case "webp": case "webp":
return FormatWebP, nil return FormatWebP
case string(FormatAVIF): case string(FormatAVIF):
return FormatAVIF, nil return FormatAVIF
case string(FormatJXL): case string(FormatJXL):
return FormatJXL, nil return FormatJXL
default: default:
return "", fmt.Errorf("%w: %s", ErrUnsupportedInputFormat, format) return FormatJPEG
} }
} }
@@ -1,39 +0,0 @@
package imageprocessor
import (
"bytes"
"errors"
"testing"
"github.com/davidbyttow/govips/v2/vips"
)
// TestImageProcessor_OrigOfUnnamedFormatRefused verifies that orig of a TIFF
// input, which is not a source format, is refused with
// ErrUnsupportedInputFormat before it is decoded, and not saved as JPEG.
func TestImageProcessor_OrigOfUnnamedFormatRefused(t *testing.T) {
t.Parallel()
img, err := vips.NewImageFromBuffer(createTestPNG(t, 20, 20))
if err != nil {
t.Fatalf("failed to decode test PNG: %v", err)
}
t.Cleanup(img.Close)
tiff, _, err := img.ExportTiff(vips.NewTiffExportParams())
if err != nil {
t.Fatalf("failed to save test TIFF: %v", err)
}
result, err := New(Params{}).Process(
t.Context(), bytes.NewReader(tiff), &Request{Format: FormatOriginal},
)
if result != nil {
t.Logf("Process() returned %s", result.ContentType)
}
if !errors.Is(err, ErrUnsupportedInputFormat) {
t.Errorf("Process() error = %v, want %v", err, ErrUnsupportedInputFormat)
}
}
@@ -1,46 +0,0 @@
package imageprocessor
import (
"bytes"
"io"
"testing"
)
// TestImageProcessor_LoadsEachSourceFormat verifies that an image in each of
// the six source formats, saved by the processor itself, is loaded and saved
// again in its own format (orig).
func TestImageProcessor_LoadsEachSourceFormat(t *testing.T) {
t.Parallel()
proc := New(Params{})
for _, format := range []Format{
FormatJPEG, FormatPNG, FormatGIF, FormatWebP, FormatAVIF, FormatJXL,
} {
saved, err := proc.Process(t.Context(),
bytes.NewReader(createTestPNG(t, 20, 20)), &Request{Format: format})
if err != nil {
t.Fatalf("saving the test image as %s: %v", format, err)
}
source, err := io.ReadAll(saved.Content)
if err != nil {
t.Fatalf("reading the %s image: %v", format, err)
}
result, err := proc.Process(t.Context(), bytes.NewReader(source),
&Request{Format: FormatOriginal})
if err != nil {
t.Errorf("orig of the %s image: %v", format, err)
continue
}
t.Logf("orig of the %s image: %s", format, result.ContentType)
if result.ContentType != FormatToMIME(format) {
t.Errorf("orig of the %s image is %s, want %s",
format, result.ContentType, FormatToMIME(format))
}
}
}
+19 -87
View File
@@ -11,8 +11,6 @@ import (
"io" "io"
"log/slog" "log/slog"
"path/filepath" "path/filepath"
"sync"
"sync/atomic"
"time" "time"
lru "github.com/hashicorp/golang-lru/v2" lru "github.com/hashicorp/golang-lru/v2"
@@ -35,6 +33,7 @@ const metaCacheSize = 10000
// CacheConfig holds cache configuration. // CacheConfig holds cache configuration.
type CacheConfig struct { type CacheConfig struct {
StateDir string StateDir string
CacheTTL time.Duration
NegativeTTL time.Duration NegativeTTL time.Duration
// MaxBytes is the disk cache size limit in bytes that eviction // MaxBytes is the disk cache size limit in bytes that eviction
@@ -80,27 +79,6 @@ type Cache struct {
evictionDone chan struct{} evictionDone chan struct{}
evictionCancel context.CancelFunc evictionCancel context.CancelFunc
// The pending counts: hits, misses, upstream fetches and transforms
// whose write to the database missed its request's deadline, kept
// here until they are written by the goroutine that
// StartPendingCountWrites starts. As for eviction, the channels are
// created in NewCache: pendingCountsAdded wakes that goroutine, and
// pendingCountsDone is closed when it returns. pendingCountsCancel,
// set by StartPendingCountWrites, cancels its context, which tells it
// to finish.
pendingHits atomic.Int64
pendingMisses atomic.Int64
pendingUpstreamFetches atomic.Int64
pendingUpstreamFetchBytes atomic.Int64
pendingTransforms atomic.Int64
pendingCountsAdded chan struct{}
pendingCountsDone chan struct{}
pendingCountsCancel context.CancelFunc
// Held through each write of the pending counts, and by
// readHitAndMissCounts to read them
pendingCountsWriteMutex sync.Mutex
// metaCache holds the content types of the variants most recently // metaCache holds the content types of the variants most recently
// stored or served, so a hit does not read the variant's .meta file. // stored or served, so a hit does not read the variant's .meta file.
// It never stands in for the variant file, which is always opened. // It never stands in for the variant file, which is always opened.
@@ -161,9 +139,6 @@ func newCache(
metaCache: metaCache, metaCache: metaCache,
contentLocks: newContentLock(), contentLocks: newContentLock(),
pendingCountsAdded: make(chan struct{}, 1),
pendingCountsDone: make(chan struct{}),
reconciliationPageSize: defaultReconciliationPageSize, reconciliationPageSize: defaultReconciliationPageSize,
} }
@@ -499,7 +474,12 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) { func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
var stats CacheStats var stats CacheStats
err := c.readHitAndMissCounts(ctx, &stats) // Fetch hit/miss counts from the stats table
err := c.db.QueryRowContext(ctx, `
SELECT hit_count, miss_count
FROM cache_stats WHERE id = 1
`).Scan(&stats.HitCount, &stats.MissCount)
if err != nil && !errors.Is(err, sql.ErrNoRows) { if err != nil && !errors.Is(err, sql.ErrNoRows) {
return nil, fmt.Errorf("failed to get cache stats: %w", err) return nil, fmt.Errorf("failed to get cache stats: %w", err)
} }
@@ -531,30 +511,19 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
} }
// IncrementStats counts a cache hit or miss, and an upstream fetch that read // IncrementStats counts a cache hit or miss, and an upstream fetch that read
// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment // fetchBytes bytes, as IncrementUpstreamFetch does.
// methods, it writes to the database with ctx's deadline but not its
// cancellation, so an ended request is still counted but does not wait for
// the database past its deadline; a count not written by then becomes a
// pending count (see StartPendingCountWrites).
func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) { func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64) {
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
defer cancel()
var err error var err error
pendingCount := &c.pendingMisses
if hit { if hit {
pendingCount = &c.pendingHits _, err = c.db.ExecContext(ctx, `
_, err = c.db.ExecContext(countCtx, `
UPDATE cache_stats UPDATE cache_stats
SET hit_count = hit_count + 1, SET hit_count = hit_count + 1,
last_updated_at = CURRENT_TIMESTAMP last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1 WHERE id = 1
`) `)
} else { } else {
_, err = c.db.ExecContext(countCtx, ` _, err = c.db.ExecContext(ctx, `
UPDATE cache_stats UPDATE cache_stats
SET miss_count = miss_count + 1, SET miss_count = miss_count + 1,
last_updated_at = CURRENT_TIMESTAMP last_updated_at = CURRENT_TIMESTAMP
@@ -562,10 +531,7 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
`) `)
} }
switch { if err != nil {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(pendingCount, 1)
case err != nil:
c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err) c.log.Warn("failed to count cache hit or miss", "hit", hit, "error", err)
} }
@@ -579,22 +545,14 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
return return
} }
countCtx, cancel := withoutCancelKeepingDeadline(ctx) _, err := c.db.ExecContext(ctx, `
defer cancel()
_, err := c.db.ExecContext(countCtx, `
UPDATE cache_stats UPDATE cache_stats
SET upstream_fetch_count = upstream_fetch_count + 1, SET upstream_fetch_count = upstream_fetch_count + 1,
upstream_fetch_bytes = upstream_fetch_bytes + ?, upstream_fetch_bytes = upstream_fetch_bytes + ?,
last_updated_at = CURRENT_TIMESTAMP last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1 WHERE id = 1
`, fetchBytes) `, fetchBytes)
if err != nil {
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(&c.pendingUpstreamFetches, 1)
c.addPendingCount(&c.pendingUpstreamFetchBytes, fetchBytes)
case err != nil:
c.log.Warn("failed to count upstream fetch", c.log.Warn("failed to count upstream fetch",
"fetch_bytes", fetchBytes, "error", err) "fetch_bytes", fetchBytes, "error", err)
} }
@@ -602,51 +560,25 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
// IncrementTransformCount counts one image transcoded by the image processor. // IncrementTransformCount counts one image transcoded by the image processor.
func (c *Cache) IncrementTransformCount(ctx context.Context) { func (c *Cache) IncrementTransformCount(ctx context.Context) {
countCtx, cancel := withoutCancelKeepingDeadline(ctx) _, err := c.db.ExecContext(ctx, `
defer cancel()
_, err := c.db.ExecContext(countCtx, `
UPDATE cache_stats UPDATE cache_stats
SET transform_count = transform_count + 1, SET transform_count = transform_count + 1,
last_updated_at = CURRENT_TIMESTAMP last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1 WHERE id = 1
`) `)
if err != nil {
switch {
case errors.Is(err, context.DeadlineExceeded):
c.addPendingCount(&c.pendingTransforms, 1)
case err != nil:
c.log.Warn("failed to count transform", "error", err) c.log.Warn("failed to count transform", "error", err)
} }
} }
// readHitAndMissCounts sets the hit and miss counts of stats to those of the
// cache_stats row plus the pending hits and misses (see
// StartPendingCountWrites). It holds pendingCountsWriteMutex, so that no
// write of the pending counts falls between the two reads.
func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) error {
c.pendingCountsWriteMutex.Lock()
defer c.pendingCountsWriteMutex.Unlock()
err := c.db.QueryRowContext(ctx, `
SELECT hit_count, miss_count
FROM cache_stats WHERE id = 1
`).Scan(&stats.HitCount, &stats.MissCount)
stats.HitCount += c.pendingHits.Load()
stats.MissCount += c.pendingMisses.Load()
return err
}
// loadVariantWithMeta is GetVariant for a variant metaCache does not // loadVariantWithMeta is GetVariant for a variant metaCache does not
// hold: it reads the content type from the variant's .meta file and // hold: it reads the content type from the variant's .meta file and
// keeps it in metaCache, unless a StoreVariant has put one there // 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, // meanwhile, as the store's is newer. A read that finds no .meta file,
// as one can when the store's .meta write failed or an eviction removed // as one can between a store's writing of the variant file and of its
// the file after the read opened the variant, serves // .meta file, serves application/octet-stream and keeps nothing, so
// application/octet-stream and keeps nothing, so metaCache only ever // metaCache only ever holds a type read from a .meta file or passed to
// holds a type read from a .meta file or passed to StoreVariant. // StoreVariant.
func (c *Cache) loadVariantWithMeta( func (c *Cache) loadVariantWithMeta(
cacheKey VariantKey, cacheKey VariantKey,
) (io.ReadCloser, int64, string, error) { ) (io.ReadCloser, int64, string, error) {
+2
View File
@@ -102,6 +102,7 @@ func setupTestCache(t *testing.T) *Cache {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
@@ -240,6 +241,7 @@ func TestCache_NegativeCacheExpiry(t *testing.T) {
// Very short negative TTL for testing // Very short negative TTL for testing
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 1 * time.Millisecond, NegativeTTL: 1 * time.Millisecond,
}) })
if err != nil { if err != nil {
+46 -89
View File
@@ -4,7 +4,6 @@ import (
"context" "context"
"database/sql" "database/sql"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io/fs" "io/fs"
"os" "os"
@@ -543,11 +542,10 @@ func (c *Cache) runReconciliationPass(ctx context.Context) {
// (or whose accounting insert failed, e.g. StoreVariant's best-effort // (or whose accounting insert failed, e.g. StoreVariant's best-effort
// insert under transient DB contention), drops accounting rows whose // insert under transient DB contention), drops accounting rows whose
// files are missing, removes source blob files the database does not // files are missing, removes source blob files the database does not
// know (and rows whose files are gone), removes source and variant files // know (and rows whose files are gone), sweeps stale temp files left
// that are empty or not the size their rows record, sweeps stale temp // behind by crashed writes, and last checks the total cache usage
// files left behind by crashed writes, and last checks the total cache // against the tables. It reads the tables a page at a time. Running it
// usage against the tables. It reads the tables a page at a time. // periodically, not just once, bounds how long such drift can
// Running it periodically, not just once, bounds how long such drift can
// accumulate unaccounted for on a long-running process to one eviction // accumulate unaccounted for on a long-running process to one eviction
// interval. Once ctx is cancelled, it stops at the next file or row and // interval. Once ctx is cancelled, it stops at the next file or row and
// returns ctx's error. // returns ctx's error.
@@ -579,9 +577,8 @@ func (c *Cache) reconcileAccounting(ctx context.Context) error {
return c.reconcileUsageTotal(ctx) return c.reconcileUsageTotal(ctx)
} }
// reconcileVariantFiles walks the variant storage directory, checking // reconcileVariantFiles walks the variant storage directory, adopting
// each variant file against its accounting row (reconcileVariantFile) // files without accounting rows and sweeping stale temp files.
// and sweeping stale temp files.
func (c *Cache) reconcileVariantFiles(ctx context.Context) error { func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
return filepath.WalkDir( return filepath.WalkDir(
c.variants.baseDir, c.variants.baseDir,
@@ -606,55 +603,38 @@ func (c *Cache) reconcileVariantFiles(ctx context.Context) error {
return nil return nil
} }
return c.reconcileVariantFile(ctx, path, entry, VariantKey(name)) return c.adoptVariantFile(ctx, path, entry, VariantKey(name))
}, },
) )
} }
// reconcileVariantFile checks one variant file against its accounting
// row. A file without a row is adopted into the accounting, unless it is
// empty. A file that is empty, or not the size its row records, as a
// crash or a restored snapshot can leave it, is removed with its row and
// a warning naming it, so the variant is converted again.
func (c *Cache) reconcileVariantFile(
ctx context.Context, path string, entry fs.DirEntry, cacheKey VariantKey,
) error {
info, err := entry.Info()
if err != nil {
return fmt.Errorf("failed to stat variant file: %w", err)
}
var recordedSize int64
err = c.db.QueryRowContext(ctx,
`SELECT size_bytes FROM variant_content WHERE cache_key = ?`, string(cacheKey),
).Scan(&recordedSize)
switch {
case errors.Is(err, sql.ErrNoRows) && info.Size() > 0:
return c.adoptVariantFile(ctx, path, info, cacheKey)
case err != nil && !errors.Is(err, sql.ErrNoRows):
return fmt.Errorf("failed to check variant accounting row: %w", err)
case info.Size() > 0 && info.Size() == recordedSize:
return nil
}
// The file is empty, or not the size its row records
c.log.Warn("removing damaged cached variant file", "path", path,
"file_bytes", info.Size(), "recorded_bytes", recordedSize)
return c.evictVariant(ctx, cacheKey)
}
// adoptVariantFile inserts an accounting row for a variant file that // adoptVariantFile inserts an accounting row for a variant file that
// has none, using the file's size and modification time. // has none, using the file's size and modification time.
func (c *Cache) adoptVariantFile( func (c *Cache) adoptVariantFile(
ctx context.Context, path string, info fs.FileInfo, cacheKey VariantKey, ctx context.Context, path string, entry fs.DirEntry, cacheKey VariantKey,
) error { ) error {
var rowExists int
err := c.db.QueryRowContext(ctx,
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`, string(cacheKey),
).Scan(&rowExists)
if err != nil {
return fmt.Errorf("failed to check variant accounting row: %w", err)
}
if rowExists > 0 {
return nil
}
info, err := entry.Info()
if err != nil {
return fmt.Errorf("failed to stat variant file: %w", err)
}
modTime := info.ModTime().UTC().Format(sqliteTimestampLayout) modTime := info.ModTime().UTC().Format(sqliteTimestampLayout)
contentType := c.variantContentTypeFromSidecar(path) contentType := c.variantContentTypeFromSidecar(path)
_, err := c.db.ExecContext(ctx, ` _, err = c.db.ExecContext(ctx, `
INSERT INTO variant_content INSERT INTO variant_content
(cache_key, size_bytes, content_type, created_at, last_accessed_at) (cache_key, size_bytes, content_type, created_at, last_accessed_at)
VALUES (?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?)
@@ -773,9 +753,10 @@ func queryStringColumn[T ~string](
return values, nil return values, nil
} }
// reconcileSourceFiles walks the source content directory, checking // reconcileSourceFiles walks the source content directory, removing
// each blob file against its source_content row (reconcileSourceFile) // blob files the database does not track (they are unreachable: source
// and sweeping stale temp files. // lookups always go through source_metadata) and sweeping stale temp
// files.
func (c *Cache) reconcileSourceFiles(ctx context.Context) error { func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
return filepath.WalkDir( return filepath.WalkDir(
c.srcContent.baseDir, c.srcContent.baseDir,
@@ -796,55 +777,31 @@ func (c *Cache) reconcileSourceFiles(ctx context.Context) error {
return nil return nil
} }
return c.reconcileSourceFile(ctx, path, entry, ContentHash(name)) return c.removeUntrackedSourceFile(ctx, path, ContentHash(name))
}, },
) )
} }
// reconcileSourceFile checks one source blob file against its
// source_content row. A file without a row is removed: it is
// unreachable, as source lookups always go through source_metadata. A
// file that is empty, or not the size its row records, as a crash or a
// restored snapshot can leave it, is removed with its rows and a warning
// naming it, so the source is fetched again.
func (c *Cache) reconcileSourceFile(
ctx context.Context, path string, entry fs.DirEntry, contentHash ContentHash,
) error {
info, err := entry.Info()
if err != nil {
return fmt.Errorf("failed to stat source content file: %w", err)
}
var recordedSize int64
err = c.db.QueryRowContext(ctx,
`SELECT size_bytes FROM source_content WHERE content_hash = ?`,
string(contentHash),
).Scan(&recordedSize)
switch {
case errors.Is(err, sql.ErrNoRows):
return c.removeUntrackedSourceFile(ctx, path, contentHash)
case err != nil:
return fmt.Errorf("failed to check source content row: %w", err)
case info.Size() > 0 && info.Size() == recordedSize:
return nil
}
// The file is empty, or not the size its row records
c.log.Warn("removing damaged cached source file", "path", path,
"file_bytes", info.Size(), "recorded_bytes", recordedSize)
return c.evictSourceBlob(ctx, contentHash)
}
// removeUntrackedSourceFile deletes a source blob file that has no // removeUntrackedSourceFile deletes a source blob file that has no
// source_content row. Any source_metadata rows referencing the hash // source_content row. Any source_metadata rows referencing the hash
// are removed first so no row ever points at a deleted file. // are removed first so no row ever points at a deleted file.
func (c *Cache) removeUntrackedSourceFile( func (c *Cache) removeUntrackedSourceFile(
ctx context.Context, path string, contentHash ContentHash, ctx context.Context, path string, contentHash ContentHash,
) error { ) error {
_, err := c.db.ExecContext(ctx, var rowExists int
err := c.db.QueryRowContext(ctx,
`SELECT COUNT(*) FROM source_content WHERE content_hash = ?`, string(contentHash),
).Scan(&rowExists)
if err != nil {
return fmt.Errorf("failed to check source content row: %w", err)
}
if rowExists > 0 {
return nil
}
_, err = c.db.ExecContext(ctx,
`DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash)) `DELETE FROM source_metadata WHERE content_hash = ?`, string(contentHash))
if err != nil { if err != nil {
return fmt.Errorf("failed to delete metadata rows for untracked blob: %w", err) return fmt.Errorf("failed to delete metadata rows for untracked blob: %w", err)
@@ -70,6 +70,7 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
// backwards compatibility with existing fixtures. // backwards compatibility with existing fixtures.
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
MaxBytes: maxBytes, MaxBytes: maxBytes,
DisableDiskCache: maxBytes == 0, DisableDiskCache: maxBytes == 0,
+1
View File
@@ -4,5 +4,6 @@ import "time"
// CacheConfig defaults. // CacheConfig defaults.
const ( const (
DefaultCacheTTL = 24 * time.Hour
DefaultNegativeTTL = 5 * time.Minute DefaultNegativeTTL = 5 * time.Minute
) )
@@ -15,6 +15,7 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: dir, StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
@@ -62,6 +63,7 @@ func TestNegativeCache_Expired(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: dir, StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 1 * time.Millisecond, // very short TTL NegativeTTL: 1 * time.Millisecond, // very short TTL
}) })
if err != nil { if err != nil {
@@ -1,359 +0,0 @@
package imgcache
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"io"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
)
// getImage calls Get for req and returns the image it answered with and
// its cache status.
func getImage(t *testing.T, svc *Service, req *ImageRequest) ([]byte, CacheStatus) {
t.Helper()
resp, err := svc.Get(t.Context(), req)
if err != nil {
t.Fatalf("Get() of quality %d error = %v", req.Quality, err)
}
defer func() { _ = resp.Content.Close() }()
body, err := io.ReadAll(resp.Content)
if err != nil {
t.Fatalf("reading the image of quality %d: %v", req.Quality, err)
}
return body, resp.CacheStatus
}
// fileSize returns the size of the file at path.
func fileSize(t *testing.T, path string) int64 {
t.Helper()
info, err := os.Stat(path)
if err != nil {
t.Fatalf("stat %s: %v", path, err)
}
return info.Size()
}
// truncateFile cuts the file at path down to size bytes.
func truncateFile(t *testing.T, path string, size int64) {
t.Helper()
t.Logf("cutting %s from %d to %d bytes",
filepath.Base(path), fileSize(t, path), size)
err := os.Truncate(path, size)
if err != nil {
t.Fatalf("truncating %s: %v", path, err)
}
}
// cachedSourcePath returns the path of the cached source file req uses.
func cachedSourcePath(t *testing.T, svc *Service, req *ImageRequest) string {
t.Helper()
contentHash, _, err := svc.cache.LookupSource(t.Context(), req)
if err != nil || contentHash == "" {
t.Fatalf("LookupSource() = %q, %v; want the cached source", contentHash, err)
}
return svc.cache.srcContent.hashToPath(contentHash)
}
// reconcileAsAtStartup runs the cache's reconciliation pass, as pixa does
// when it starts, and returns what the pass logged.
func reconcileAsAtStartup(t *testing.T, cache *Cache) string {
t.Helper()
var logBuf bytes.Buffer
cache.log = slog.New(slog.NewJSONHandler(&logBuf, nil))
err := cache.reconcileAccounting(t.Context())
if err != nil {
t.Fatalf("reconcileAccounting() error = %v", err)
}
t.Logf("reconciliation log: %s", logBuf.String())
return logBuf.String()
}
// assertWarningNames checks that logs hold a warning naming the file at
// path.
func assertWarningNames(t *testing.T, logs, path string) {
t.Helper()
for line := range strings.Lines(logs) {
if strings.Contains(line, `"level":"WARN"`) &&
strings.Contains(line, `"path":"`+path+`"`) {
return
}
}
t.Errorf("reconciliation logged no warning naming %s", path)
}
// TestService_Get_FetchesAgainForEmptyCachedSource empties the cached source
// file, as a crash or a restored snapshot could, and runs the reconciliation
// pass, which must remove it with a warning naming it. It then asks for two
// new sizes of the source: the first must fetch it again and cache it whole,
// so that the second is made from the cache without a third fetch.
func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
getImage(t, svc, photoVariant(fixtures, 85, FitCover))
path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, FitCover))
wholeSize := fileSize(t, path)
truncateFile(t, path, 0)
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
getImage(t, svc, photoVariant(fixtures, 40, FitCover))
getImage(t, svc, photoVariant(fixtures, 30, FitCover))
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
t.Logf("upstream fetches = %d", fetches)
if fetches != 2 {
t.Errorf("upstream fetches = %d, want 2: the empty cached source "+
"was kept and fetched again on every new size", fetches)
}
size := fileSize(t, path)
if size != wholeSize {
t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
}
}
// TestService_Get_FetchesAgainForPartialCachedSource cuts the cached source
// file to half its size, as a crash or a restored snapshot could, and runs
// the reconciliation pass, which must remove it with a warning naming it.
// It then asks for two new sizes of the source: the first must fetch it
// again rather than fail to decode the partial file, and the second is
// made from the cache.
func TestService_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
getImage(t, svc, photoVariant(fixtures, 85, FitCover))
path := cachedSourcePath(t, svc, photoVariant(fixtures, 85, FitCover))
wholeSize := fileSize(t, path)
truncateFile(t, path, wholeSize/2)
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
getImage(t, svc, photoVariant(fixtures, 40, FitCover))
getImage(t, svc, photoVariant(fixtures, 30, FitCover))
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
t.Logf("upstream fetches = %d", fetches)
if fetches != 2 {
t.Errorf("upstream fetches = %d, want 2", fetches)
}
size := fileSize(t, path)
if size != wholeSize {
t.Errorf("cached source is %d bytes, want %d", size, wholeSize)
}
}
// assertConvertedAgain caches a variant of the photo, cuts its file to size
// bytes and runs the reconciliation pass, which must remove it with a
// warning naming it. It then checks that Get converts the image again
// instead of serving the file, and that the next Get serves the whole
// variant from the cache.
func assertConvertedAgain(t *testing.T, size int64) {
t.Helper()
svc, fixtures := SetupTestService(t)
req := photoVariant(fixtures, 85, FitCover)
want, _ := getImage(t, svc, req)
path := svc.cache.variants.keyToPath(CacheKey(req))
truncateFile(t, path, size)
assertWarningNames(t, reconcileAsAtStartup(t, svc.cache), path)
got, status := getImage(t, svc, req)
t.Logf("after the cut: %d bytes, %s", len(got), status)
if status != CacheMiss || !bytes.Equal(got, want) {
t.Errorf("after the cut: Get() = %d bytes, %s; want the %d bytes "+
"converted again, %s", len(got), status, len(want), CacheMiss)
}
got, status = getImage(t, svc, req)
t.Logf("next request: %d bytes, %s", len(got), status)
if status != CacheHit || !bytes.Equal(got, want) {
t.Errorf("next request: Get() = %d bytes, %s; want the %d bytes, %s",
len(got), status, len(want), CacheHit)
}
}
// TestService_Get_ConvertsAgainForEmptyCachedVariant empties a cached
// variant file, as a crash or a restored snapshot could, and checks that it
// is converted again rather than served empty.
func TestService_Get_ConvertsAgainForEmptyCachedVariant(t *testing.T) {
t.Parallel()
assertConvertedAgain(t, 0)
}
// TestService_Get_ConvertsAgainForPartialCachedVariant cuts a cached
// variant file to ten bytes, as a crash or a restored snapshot could, and
// checks that it is converted again rather than served cut short.
func TestService_Get_ConvertsAgainForPartialCachedVariant(t *testing.T) {
t.Parallel()
assertConvertedAgain(t, 10)
}
// TestReconciliationRemovesEmptyVariantFileWithoutRow checks that the
// reconciliation pass removes an empty variant file that has no
// accounting row, with a warning naming it, instead of adding a row for
// it that would have it served empty.
func TestReconciliationRemovesEmptyVariantFileWithoutRow(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
_, err := cache.variants.Store(testVariantKeyOne, bytes.NewReader(nil),
testContentTypeWebP)
if err != nil {
t.Fatalf("failed to store an empty variant file: %v", err)
}
logs := reconcileAsAtStartup(t, cache)
assertWarningNames(t, logs, cache.variants.keyToPath(testVariantKeyOne))
if cache.variants.Exists(testVariantKeyOne) {
t.Error("the empty variant file is still on disk")
}
rows := countRows(t, cache,
`SELECT COUNT(*) FROM variant_content WHERE cache_key = ?`,
string(testVariantKeyOne))
if rows != 0 {
t.Errorf("the empty variant file got %d accounting rows, want 0", rows)
}
}
// TestVariantStorage_StoreReplacesMetaFile stores a variant over one
// already stored and checks that Store puts a new .meta file in place of
// the old one, as it does the image, rather than writing into the old
// file, which a crash could leave empty. The old .meta file is given a
// second name before the store, so that it can be told apart from the new
// one after it.
func TestVariantStorage_StoreReplacesMetaFile(t *testing.T) {
t.Parallel()
storage, key := storeTestVariant(t)
metaPath := storage.keyToPath(key) + variantMetaSuffix
oldMetaPath := metaPath + ".old"
err := os.Link(metaPath, oldMetaPath)
if err != nil {
t.Fatalf("giving the .meta file a second name: %v", err)
}
_, err = storage.Store(key, bytes.NewReader([]byte("new variant data")),
testContentTypeJPEG)
if err != nil {
t.Fatalf("Store() error = %v", err)
}
oldMeta, err := os.Stat(oldMetaPath)
if err != nil {
t.Fatalf("stat of the old .meta file: %v", err)
}
newMeta, err := os.Stat(metaPath)
if err != nil {
t.Fatalf("stat of the new .meta file: %v", err)
}
if os.SameFile(oldMeta, newMeta) {
t.Error("Store() wrote into the old .meta file instead of putting " +
"a new one in its place")
}
reader, _, contentType, err := storage.LoadWithMeta(key)
if err != nil {
t.Fatalf("LoadWithMeta() error = %v", err)
}
_ = reader.Close()
t.Logf("content type after the store: %s", contentType)
if contentType != testContentTypeJPEG {
t.Errorf("content type = %q, want %q", contentType, testContentTypeJPEG)
}
}
// TestContentStorage_StoreHashedReplacesFileOfAnotherSize checks that a
// store over a file of another size at the content's path, such as a
// partial file a crash left, replaces it with the content.
func TestContentStorage_StoreHashedReplacesFileOfAnotherSize(t *testing.T) {
t.Parallel()
storage, err := NewContentStorage(t.TempDir())
if err != nil {
t.Fatalf("NewContentStorage() error = %v", err)
}
content := []byte("the whole source image")
sum := sha256.Sum256(content)
hash := ContentHash(hex.EncodeToString(sum[:]))
path := storage.hashToPath(hash)
err = os.MkdirAll(filepath.Dir(path), StorageDirPerm)
if err != nil {
t.Fatalf("creating the content's directory: %v", err)
}
err = os.WriteFile(path, content[:5], StorageFilePerm)
if err != nil {
t.Fatalf("writing a partial file: %v", err)
}
_, err = storage.StoreHashed(hash, content)
if err != nil {
t.Fatalf("StoreHashed() error = %v", err)
}
stored, err := storage.Load(hash)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
defer func() { _ = stored.Close() }()
got, err := io.ReadAll(stored)
if err != nil {
t.Fatalf("reading the stored content: %v", err)
}
if !bytes.Equal(got, content) {
t.Errorf("stored file = %q, want %q", got, content)
}
}
-149
View File
@@ -1,149 +0,0 @@
package imgcache
import (
"context"
"fmt"
"sync/atomic"
)
// StartPendingCountWrites starts the goroutine that writes the pending
// counts to the database whenever there are some, one UPDATE at a time:
// however many requests pass their deadline before their counts are
// written, at most this one write waits for the database. It is a no-op
// when already started. The goroutine outlives the caller, so it runs with
// its own context, which StopPendingCountWrites cancels to tell it to
// finish.
func (c *Cache) StartPendingCountWrites() {
if c.pendingCountsCancel != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
c.pendingCountsCancel = cancel
go func() {
c.pendingCountWriteLoop(ctx)
}()
}
// StopPendingCountWrites tells the goroutine StartPendingCountWrites
// started, if it did, to finish, and waits for its write under way to end
// and for it to return. It then writes the pending counts left, so that
// they are written before the database closes. It waits for all of this at
// most until ctx ends, then logs the counts not written and returns an
// error.
func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
var err error
if c.pendingCountsCancel != nil {
c.pendingCountsCancel()
select {
case <-c.pendingCountsDone:
case <-ctx.Done():
// The write given up on here may not reach the database:
// database/sql closes without waiting for a query, and a
// write still waiting for the connection fails when the
// database closes.
err = fmt.Errorf("pending counts still being written: %w", ctx.Err())
}
}
if err == nil {
err = c.writePendingCounts(ctx)
}
if err != nil {
c.log.Error("counts not written at shutdown",
"hits", c.pendingHits.Load(),
"misses", c.pendingMisses.Load(),
"upstream_fetches", c.pendingUpstreamFetches.Load(),
"upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(),
"transforms", c.pendingTransforms.Load(),
"error", err,
)
return err
}
return nil
}
// pendingCountWriteLoop is the body of the goroutine
// StartPendingCountWrites starts. Once ctx is cancelled it starts no write
// and returns, but does not cut off a write under way then. A write that
// fails leaves the counts pending, for the next write.
func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
defer close(c.pendingCountsDone)
for {
select {
case <-ctx.Done():
return
case <-c.pendingCountsAdded:
}
// select picks at random when ctx is cancelled and a wakeup
// waits too.
if ctx.Err() != nil {
return
}
err := c.writePendingCounts(context.WithoutCancel(ctx))
if err != nil {
c.log.Warn("failed to write pending counts", "error", err)
}
}
}
// addPendingCount adds n to pendingCount, one of the pending counts, and
// wakes the goroutine that writes them. pendingCountsAdded has capacity one,
// so a wakeup already waiting is enough.
func (c *Cache) addPendingCount(pendingCount *atomic.Int64, n int64) {
pendingCount.Add(n)
select {
case c.pendingCountsAdded <- struct{}{}:
default:
}
}
// writePendingCounts adds the pending counts to the cache_stats row in one
// UPDATE, then takes what it wrote out of them, leaving any added
// meanwhile. When the UPDATE fails, they all stay pending.
func (c *Cache) writePendingCounts(ctx context.Context) error {
c.pendingCountsWriteMutex.Lock()
defer c.pendingCountsWriteMutex.Unlock()
hits := c.pendingHits.Load()
misses := c.pendingMisses.Load()
upstreamFetches := c.pendingUpstreamFetches.Load()
upstreamFetchBytes := c.pendingUpstreamFetchBytes.Load()
transforms := c.pendingTransforms.Load()
if hits+misses+upstreamFetches+upstreamFetchBytes+transforms == 0 {
return nil
}
_, err := c.db.ExecContext(ctx, `
UPDATE cache_stats
SET hit_count = hit_count + ?,
miss_count = miss_count + ?,
upstream_fetch_count = upstream_fetch_count + ?,
upstream_fetch_bytes = upstream_fetch_bytes + ?,
transform_count = transform_count + ?,
last_updated_at = CURRENT_TIMESTAMP
WHERE id = 1
`, hits, misses, upstreamFetches, upstreamFetchBytes, transforms)
if err != nil {
return fmt.Errorf("failed to write pending counts: %w", err)
}
c.pendingHits.Add(-hits)
c.pendingMisses.Add(-misses)
c.pendingUpstreamFetches.Add(-upstreamFetches)
c.pendingUpstreamFetchBytes.Add(-upstreamFetchBytes)
c.pendingTransforms.Add(-transforms)
return nil
}
@@ -1,327 +0,0 @@
package imgcache
import (
"context"
"database/sql"
"errors"
"testing"
"time"
)
// pendingCountsWait is how long a test waits for the database's connection,
// for a count to reach the database, for a write to start waiting for the
// connection, for a request to return past its deadline, or for
// StopPendingCountWrites to return.
const pendingCountsWait = 5 * time.Second
// otherWritesWait is how long a test gives other writes, once one write waits
// for the database's connection, to start waiting for it too.
const otherWritesWait = 100 * time.Millisecond
// holdDatabase takes the one connection of cache's database, so that every
// other query waits for it, and returns the func that frees it.
func holdDatabase(t *testing.T, cache *Cache) func() {
t.Helper()
ctx, cancel := context.WithTimeout(t.Context(), pendingCountsWait)
defer cancel()
conn, err := cache.db.Conn(ctx)
if err != nil {
t.Fatalf("failed to take the database connection: %v", err)
}
return func() { _ = conn.Close() }
}
// waitForConnectionWaits waits until db.Stats().WaitCount, the number of
// times a caller has waited for a connection, reaches waits.
func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) {
t.Helper()
deadline := time.Now().Add(pendingCountsWait)
for db.Stats().WaitCount < waits {
if time.Now().After(deadline) {
t.Fatalf("callers waited for the database connection %d times, want %d",
db.Stats().WaitCount, waits)
}
time.Sleep(10 * time.Millisecond)
}
}
// waitForCounters waits until the cache_stats row holds want.
func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) {
t.Helper()
deadline := time.Now().Add(pendingCountsWait)
for {
got := readCacheStatsCounters(t, cache)
if got == want {
return
}
if time.Now().After(deadline) {
t.Fatalf("counters = %+v, want %+v", got, want)
}
time.Sleep(10 * time.Millisecond)
}
}
// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
// database's one connection while a request whose fetch is held reaches its
// deadline. The request must return with the deadline's error while the
// connection is still held, as it waits for no database write past its
// deadline, and its miss must reach the database once the connection is free.
func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
t.Parallel()
svc, fixtures, fetcher := setupHeldFetchService(t)
svc.cache.StartPendingCountWrites()
defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
// Room for the request to reach its fetch on a busy host
const timeout = 2 * time.Second
ctx, cancel := context.WithTimeout(t.Context(), timeout)
defer cancel()
deadline, _ := ctx.Deadline()
results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover))
// The request has made its database reads by the time it fetches.
select {
case <-fetcher.started:
case <-time.After(timeout):
t.Fatal("request did not reach its fetch by its deadline")
}
releaseDatabase := holdDatabase(t, svc.cache)
defer releaseDatabase()
select {
case got := <-results:
t.Logf("Get() returned %v after its deadline, error = %v",
time.Since(deadline), got.err)
if !errors.Is(got.err, context.DeadlineExceeded) {
t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
}
case <-time.After(time.Until(deadline) + pendingCountsWait):
t.Fatal("request did not return while the database was busy")
}
releaseDatabase()
waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1})
}
// TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting holds the
// database's one connection while many misses are counted past their
// deadline. At most one write may then wait for the connection, and every
// miss must reach the database once it is free.
func TestIncrementStats_ManyCountsPastTheirDeadlineLeaveOneWriteWaiting(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
cache.StartPendingCountWrites()
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
const misses = 20
for range misses {
cache.IncrementStats(ended, false, 0)
}
// Once one write waits for the connection, any others start waiting too.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
time.Sleep(otherWritesWait)
waits := cache.db.Stats().WaitCount - waitsBefore
t.Logf("%d writes waited for the database", waits)
if waits > 1 {
t.Errorf("%d writes waited for the database, want at most 1", waits)
}
releaseDatabase()
waitForCounters(t, cache, cacheStatsCounters{missCount: misses})
}
// TestStats_IncludesPendingCounts counts hits and misses whose writes miss
// their deadline, and checks that Stats adds them to the database's counts
// before they are written.
func TestStats_IncludesPendingCounts(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
_, err := cache.db.ExecContext(t.Context(),
`UPDATE cache_stats SET hit_count = 75, miss_count = 25 WHERE id = 1`)
if err != nil {
t.Fatal(err)
}
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(ended, true, 0)
cache.IncrementStats(ended, false, 0)
stats, err := cache.Stats(t.Context())
if err != nil {
t.Fatalf("Stats() error = %v", err)
}
if stats.HitCount != 77 || stats.MissCount != 26 {
t.Errorf("HitCount = %d, MissCount = %d, want 77 and 26",
stats.HitCount, stats.MissCount)
}
want := cacheStatsCounters{hitCount: 75, missCount: 25}
if got := readCacheStatsCounters(t, cache); got != want {
t.Errorf("counters in the database = %+v, want %+v", got, want)
}
}
// TestStopPendingCountWrites_WritesPendingCounts holds the database's one
// connection while the goroutine that writes the pending counts waits for it,
// counts more, then stops that goroutine. The stop must leave the write under
// way to finish rather than cut it off, and once the connection is free it
// must write the counts that write left.
func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
cache.StartPendingCountWrites()
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
// The goroutine's write waits for the connection.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
// Counted after that write read the pending counts
cache.IncrementStats(ended, false, 0)
cache.IncrementUpstreamFetch(ended, 1024)
cache.IncrementTransformCount(ended)
// The goroutine is told to finish here as well as by the stop, so that it
// is told before its write under way ends, however late the stop runs.
// The counts above then reach the database only through the stop's own
// write.
cache.pendingCountsCancel()
stopped := make(chan error, 1)
go func() {
stopped <- cache.StopPendingCountWrites(t.Context())
}()
// Had the stop cut off the write under way, its own write would wait for
// the connection too.
time.Sleep(otherWritesWait)
waits := cache.db.Stats().WaitCount - waitsBefore
t.Logf("%d writes waited for the database", waits)
if waits > 1 {
t.Errorf("%d writes waited for the database, want 1: "+
"the stop cut off the write under way", waits)
}
releaseDatabase()
select {
case err := <-stopped:
if err != nil {
t.Fatalf("StopPendingCountWrites() error = %v", err)
}
case <-time.After(pendingCountsWait):
t.Fatal("StopPendingCountWrites() did not return once the database was free")
}
want := cacheStatsCounters{1, 1, 1, 1024, 1}
if got := readCacheStatsCounters(t, cache); got != want {
t.Errorf("counters = %+v, want %+v", got, want)
}
}
// TestStopPendingCountWrites_ReturnsWhenItsContextEnds holds the database's
// one connection while the goroutine that writes the pending counts waits for
// it, then stops that goroutine with a context that ends first. The stop must
// return the context's error, and the write under way must still reach the
// database once the connection is free.
func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
t.Parallel()
cache, _ := newEvictionTestCache(t, 1<<30)
cache.StartPendingCountWrites()
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
releaseDatabase := holdDatabase(t, cache)
defer releaseDatabase()
waitsBefore := cache.db.Stats().WaitCount
// Writes given a context past its deadline fail at once.
ended, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel()
cache.IncrementStats(ended, true, 0)
// The goroutine's write waits for the connection.
waitForConnectionWaits(t, cache.db, waitsBefore+1)
stopCtx, cancelStop := context.WithCancel(t.Context())
cancelStop()
stopped := make(chan error, 1)
go func() {
stopped <- cache.StopPendingCountWrites(stopCtx)
}()
select {
case err := <-stopped:
if !errors.Is(err, context.Canceled) {
t.Errorf("StopPendingCountWrites() error = %v, want %v",
err, context.Canceled)
}
case <-time.After(pendingCountsWait):
t.Fatal("StopPendingCountWrites() did not return once its context ended")
}
releaseDatabase()
waitForCounters(t, cache, cacheStatsCounters{hitCount: 1})
}
+12 -20
View File
@@ -162,7 +162,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
// Fall through to re-process // Fall through to re-process
} else { } else {
// Counted also when the request context has ended meanwhile // Counted also when the request context has ended meanwhile
s.cache.IncrementStats(ctx, true, 0) s.cache.IncrementStats(context.WithoutCancel(ctx), true, 0)
return &ImageResponse{ return &ImageResponse{
Content: reader, Content: reader,
@@ -179,7 +179,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
// failed or the request context has ended meanwhile // failed or the request context has ended meanwhile
response, err := s.processOrWait(ctx, req) response, err := s.processOrWait(ctx, req)
s.cache.IncrementStats(ctx, false, 0) s.cache.IncrementStats(context.WithoutCancel(ctx), false, 0)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -300,8 +300,14 @@ func (s *Service) processOrWait(
} }
}() }()
processingCtx, cancel := withoutCancelKeepingDeadline(ctx) processingCtx := context.WithoutCancel(ctx)
defer cancel()
if deadline, ok := ctx.Deadline(); ok {
var cancel context.CancelFunc
processingCtx, cancel = context.WithDeadline(processingCtx, deadline)
defer cancel()
}
return s.processFromSourceOrFetch(processingCtx, req, cacheKey) return s.processFromSourceOrFetch(processingCtx, req, cacheKey)
}) })
@@ -338,20 +344,6 @@ func (s *Service) processOrWait(
}, nil }, nil
} }
// withoutCancelKeepingDeadline returns ctx without its cancellation but with
// its deadline, if it has one, and the func that releases the returned
// context.
func withoutCancelKeepingDeadline(
ctx context.Context,
) (context.Context, context.CancelFunc) {
deadline, hasDeadline := ctx.Deadline()
if !hasDeadline {
return context.WithoutCancel(ctx), func() {}
}
return context.WithDeadline(context.WithoutCancel(ctx), deadline)
}
// loadCachedSource opens source content from cache, without reading it, and // loadCachedSource opens source content from cache, without reading it, and
// returns it with its size; nil if the cached data is unavailable, empty or // returns it with its size; nil if the cached data is unavailable, empty or
// exceeds maxResponseSize. // exceeds maxResponseSize.
@@ -461,7 +453,7 @@ func (s *Service) fetchAndProcess(
fetchBytes := int64(len(sourceData)) fetchBytes := int64(len(sourceData))
// Counted also when the request context has ended meanwhile // Counted also when the request context has ended meanwhile
s.cache.IncrementUpstreamFetch(ctx, fetchBytes) s.cache.IncrementUpstreamFetch(context.WithoutCancel(ctx), fetchBytes)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to read upstream response: %w", err) return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -543,7 +535,7 @@ func (s *Service) processAndStore(
processDuration := time.Since(processStart) processDuration := time.Since(processStart)
// Counted also when the request context has ended meanwhile // Counted also when the request context has ended meanwhile
s.cache.IncrementTransformCount(ctx) s.cache.IncrementTransformCount(context.WithoutCancel(ctx))
// Read processed content // Read processed content
processedData, err := io.ReadAll(processResult.Content) processedData, err := io.ReadAll(processResult.Content)
-168
View File
@@ -2,9 +2,7 @@ package imgcache
import ( import (
"context" "context"
"encoding/json"
"io" "io"
"os"
"testing" "testing"
"time" "time"
@@ -432,172 +430,6 @@ func TestService_Get_Caching(t *testing.T) {
} }
} }
// TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo caches a source and
// a variant of it and sets back two days every time pixa stored for them.
// For the source: fetched_at in source_metadata, fetched_at and
// last_accessed_at in source_content, the fetched_at in its metadata file,
// and the access and modification times of its file and of its metadata
// file. For the variant: created_at and last_accessed_at in variant_content,
// the created_at in its .meta file, and the access and modification times of
// its file and of its .meta file. It then checks that the variant is still
// served from the cache and that a new variant is made from the cached source
// without fetching it again.
//
// Left at the present: the files' status change and birth times, which a
// program cannot set, and the times of the directories holding the files,
// which other entries share. The source's stored response headers are empty,
// as the mock fetcher sends none. expires_at, etag and last_modified in
// source_metadata and in the metadata file are never written, nor are the
// request_cache and output_content tables.
func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
variant := photoVariant(fixtures, 85, FitCover)
resp, err := svc.Get(t.Context(), variant)
if err != nil {
t.Fatalf("Get() first request error = %v", err)
}
_ = resp.Content.Close()
twoDaysAgo := time.Now().AddDate(0, 0, -2)
backdateSource(t, svc.cache, variant, twoDaysAgo)
backdateVariant(t, svc.cache, variant, twoDaysAgo)
resp, err = svc.Get(t.Context(), variant)
if err != nil {
t.Fatalf("Get() of the cached variant error = %v", err)
}
_ = resp.Content.Close()
if resp.CacheStatus != CacheHit {
t.Errorf("cached variant CacheStatus = %q, want %q",
resp.CacheStatus, CacheHit)
}
// Another quality is a new variant, made from the cached source
resp, err = svc.Get(t.Context(), photoVariant(fixtures, 40, FitCover))
if err != nil {
t.Fatalf("Get() of a new variant error = %v", err)
}
_ = resp.Content.Close()
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
if fetches != 1 {
t.Errorf("upstream fetches = %d, want 1", fetches)
}
}
// backdateSource sets every time pixa stored for req's cached source to when,
// as listed on TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo.
func backdateSource(
t *testing.T, cache *Cache, req *ImageRequest, when time.Time,
) {
t.Helper()
pathHash := HashPath(req.SourcePath + "?" + req.SourceQuery)
meta, err := cache.srcMetadata.Load(req.SourceHost, pathHash)
if err != nil {
t.Fatalf("reading the source's metadata file: %v", err)
}
meta.FetchedAt = when.Unix()
err = cache.srcMetadata.Store(req.SourceHost, pathHash, meta)
if err != nil {
t.Fatalf("writing the source's metadata file: %v", err)
}
dbTime := when.UTC().Format(sqliteTimestampFormat)
updateOneRow(t, cache, `
UPDATE source_metadata SET fetched_at = ?
WHERE source_host = ? AND source_path = ? AND source_query = ?
`, dbTime, req.SourceHost, req.SourcePath, req.SourceQuery)
updateOneRow(t, cache,
`UPDATE source_content SET fetched_at = ? WHERE content_hash = ?`,
dbTime, meta.ContentHash)
setSourceLastAccessed(t, cache, ContentHash(meta.ContentHash), when)
for _, path := range []string{
cache.srcContent.hashToPath(ContentHash(meta.ContentHash)),
cache.srcMetadata.metaPath(req.SourceHost, pathHash),
} {
err = os.Chtimes(path, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", path, err)
}
}
}
// backdateVariant sets every time pixa stored for the cached variant req asks
// for to when, as listed on
// TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo.
func backdateVariant(
t *testing.T, cache *Cache, req *ImageRequest, when time.Time,
) {
t.Helper()
key := CacheKey(req)
content, size, contentType, err := cache.variants.LoadWithMeta(key)
if err != nil || contentType == "" {
t.Fatalf("reading the variant's .meta file: content type %q, error %v",
contentType, err)
}
_ = content.Close()
meta, err := json.Marshal(VariantMeta{
ContentType: contentType,
Size: size,
CreatedAt: when.Unix(),
})
if err != nil {
t.Fatalf("encoding the variant's .meta file: %v", err)
}
path := cache.variants.keyToPath(key)
err = os.WriteFile(path+variantMetaSuffix, meta, StorageFilePerm)
if err != nil {
t.Fatalf("writing the variant's .meta file: %v", err)
}
updateOneRow(t, cache,
`UPDATE variant_content SET created_at = ? WHERE cache_key = ?`,
when.UTC().Format(sqliteTimestampFormat), string(key))
setVariantLastAccessed(t, cache, key, when)
for _, file := range []string{path, path + variantMetaSuffix} {
err = os.Chtimes(file, when, when)
if err != nil {
t.Fatalf("setting back the times of %s: %v", file, err)
}
}
}
// updateOneRow runs the UPDATE statement query and fails the test unless it
// changed exactly one row.
func updateOneRow(t *testing.T, cache *Cache, query string, args ...any) {
t.Helper()
result, err := cache.db.ExecContext(t.Context(), query, args...)
if err != nil {
t.Fatalf("%s: %v", query, err)
}
rows, err := result.RowsAffected()
if err != nil || rows != 1 {
t.Fatalf("%s: rows = %d, error = %v, want 1 row", query, rows, err)
}
}
func TestService_Get_DifferentSizes(t *testing.T) { func TestService_Get_DifferentSizes(t *testing.T) {
t.Parallel() t.Parallel()
+5
View File
@@ -43,6 +43,7 @@ func TestStats_HitRateIsRatio(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: dir, StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
@@ -90,6 +91,7 @@ func TestStats_ZeroCounts(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: dir, StateDir: dir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
@@ -118,6 +120,7 @@ func TestStats_LogsFailedCountQueries(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)), Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
}) })
@@ -159,6 +162,7 @@ func TestIncrementStats_LogsFailedUpdates(t *testing.T) {
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)), Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
}) })
@@ -228,6 +232,7 @@ func TestStats_DisabledCacheReportsNoItems(t *testing.T) {
disabled, err := NewCache(enabled.db, CacheConfig{ disabled, err := NewCache(enabled.db, CacheConfig{
StateDir: t.TempDir(), StateDir: t.TempDir(),
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
DisableDiskCache: true, DisableDiskCache: true,
}) })
+60 -66
View File
@@ -79,8 +79,8 @@ func (s *ContentStorage) Store(r io.Reader) (ContentHash, int64, error) {
// from hash, without recomputing it. Callers that already know the // from hash, without recomputing it. Callers that already know the
// hash before writing (e.g. because they must hold a hash-keyed lock // hash before writing (e.g. because they must hold a hash-keyed lock
// across the whole store operation) use this instead of Store. Like // across the whole store operation) use this instead of Store. Like
// Store, it is idempotent: a file of the content's size already at that // Store, it is idempotent: content already on disk at that path is
// path is left untouched, and a file of another size is replaced. // left untouched.
func (s *ContentStorage) StoreHashed( func (s *ContentStorage) StoreHashed(
hash ContentHash, data []byte, hash ContentHash, data []byte,
) (int64, error) { ) (int64, error) {
@@ -151,35 +151,28 @@ func (s *ContentStorage) Exists(hash ContentHash) bool {
return err == nil return err == nil
} }
// writeIfAbsent writes data with writeFileSynced to the path derived // writeIfAbsent writes data to the path derived from hash, unless
// from hash, unless a file of the data's size is already there. // content already exists there, via a temp-file-plus-rename so
// concurrent readers never observe a partial file.
func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error { func (s *ContentStorage) writeIfAbsent(hash ContentHash, data []byte) error {
// Build path: <basedir>/<ab>/<cd>/<hash> // Build path: <basedir>/<ab>/<cd>/<hash>
path := s.hashToPath(hash) path := s.hashToPath(hash)
// A file of another size there is damaged, and is replaced // Check if already exists
info, statErr := os.Stat(path) _, statErr := os.Stat(path)
if statErr == nil && info.Size() == int64(len(data)) { if statErr == nil {
return nil return nil
} }
err := os.MkdirAll(filepath.Dir(path), StorageDirPerm) // Create directory structure
dir := filepath.Dir(path)
err := os.MkdirAll(dir, StorageDirPerm)
if err != nil { if err != nil {
return fmt.Errorf("failed to create directory: %w", err) return fmt.Errorf("failed to create directory: %w", err)
} }
return writeFileSynced(path, data) // Write to temp file first, then rename for atomicity
}
// writeFileSynced writes data to path through a temp file in the same
// directory: the temp file is synced to disk, renamed to path, and the
// directory synced after, so that neither a concurrent reader nor a
// crash finds a partial file under path. When the directory sync fails,
// the file is removed from path, so that a failed write leaves no file
// there.
func writeFileSynced(path string, data []byte) error {
dir := filepath.Dir(path)
tmpFile, err := os.CreateTemp(dir, ".tmp-*") tmpFile, err := os.CreateTemp(dir, ".tmp-*")
if err != nil { if err != nil {
return fmt.Errorf("failed to create temp file: %w", err) return fmt.Errorf("failed to create temp file: %w", err)
@@ -187,20 +180,18 @@ func writeFileSynced(path string, data []byte) error {
tmpPath := tmpFile.Name() tmpPath := tmpFile.Name()
// Each failure path below unlinks the temp file explicitly. This
// replaces a deferred cleanup that read a named result, which the
// canonical config does not permit; the set of paths that remove
// tmpPath, and the order relative to Close, is unchanged. This
// mirrors how MetadataStorage.Store and VariantStorage.Store below
// already express the same cleanup.
_, err = tmpFile.Write(data) _, err = tmpFile.Write(data)
if err != nil { if err != nil {
_ = tmpFile.Close() _ = tmpFile.Close()
_ = os.Remove(tmpPath) _ = os.Remove(tmpPath)
return fmt.Errorf("failed to write temp file: %w", err) return fmt.Errorf("failed to write content: %w", err)
}
err = tmpFile.Sync()
if err != nil {
_ = tmpFile.Close()
_ = os.Remove(tmpPath)
return fmt.Errorf("failed to sync temp file: %w", err)
} }
err = tmpFile.Close() err = tmpFile.Close()
@@ -210,6 +201,7 @@ func writeFileSynced(path string, data []byte) error {
return fmt.Errorf("failed to close temp file: %w", err) return fmt.Errorf("failed to close temp file: %w", err)
} }
// Atomic rename
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path)) err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
if err != nil { if err != nil {
_ = os.Remove(tmpPath) _ = os.Remove(tmpPath)
@@ -217,31 +209,6 @@ func writeFileSynced(path string, data []byte) error {
return fmt.Errorf("failed to rename temp file: %w", err) return fmt.Errorf("failed to rename temp file: %w", err)
} }
err = syncDirectory(dir)
if err != nil {
_ = os.Remove(path)
return err
}
return nil
}
// syncDirectory syncs the directory dir to disk, so that a file just
// renamed into it is still there after a crash.
func syncDirectory(dir string) error {
d, err := os.Open(filepath.Clean(dir))
if err != nil {
return fmt.Errorf("failed to open directory: %w", err)
}
err = d.Sync()
_ = d.Close()
if err != nil {
return fmt.Errorf("failed to sync directory: %w", err)
}
return nil return nil
} }
@@ -449,9 +416,7 @@ func NewVariantStorage(baseDir string, log *slog.Logger) (*VariantStorage, error
return &VariantStorage{baseDir: baseDir, log: log}, nil return &VariantStorage{baseDir: baseDir, log: log}, nil
} }
// Store writes content and metadata to storage at the given key, each // Store writes content and metadata to storage at the given key.
// with writeFileSynced. The .meta file is written first, so that a crash
// cannot leave the content in place without it.
func (s *VariantStorage) Store( func (s *VariantStorage) Store(
key VariantKey, r io.Reader, contentType string, key VariantKey, r io.Reader, contentType string,
) (int64, error) { ) (int64, error) {
@@ -464,11 +429,46 @@ func (s *VariantStorage) Store(
path := s.keyToPath(key) path := s.keyToPath(key)
metaPath := path + ".meta" metaPath := path + ".meta"
err = os.MkdirAll(filepath.Dir(path), StorageDirPerm) // Create directory structure
dir := filepath.Dir(path)
err = os.MkdirAll(dir, StorageDirPerm)
if err != nil { if err != nil {
return 0, fmt.Errorf("failed to create directory: %w", err) return 0, fmt.Errorf("failed to create directory: %w", err)
} }
// Write content to temp file first, then rename for atomicity
tmpFile, err := os.CreateTemp(dir, ".tmp-*")
if err != nil {
return 0, fmt.Errorf("failed to create temp file: %w", err)
}
tmpPath := tmpFile.Name()
_, err = tmpFile.Write(data)
if err != nil {
_ = tmpFile.Close()
_ = os.Remove(tmpPath)
return 0, fmt.Errorf("failed to write content: %w", err)
}
err = tmpFile.Close()
if err != nil {
_ = os.Remove(tmpPath)
return 0, fmt.Errorf("failed to close temp file: %w", err)
}
// Atomic rename content
err = os.Rename(filepath.Clean(tmpPath), filepath.Clean(path))
if err != nil {
_ = os.Remove(tmpPath)
return 0, fmt.Errorf("failed to rename temp file: %w", err)
}
// Write metadata
meta := VariantMeta{ meta := VariantMeta{
ContentType: contentType, ContentType: contentType,
Size: size, Size: size,
@@ -480,19 +480,13 @@ func (s *VariantStorage) Store(
return 0, fmt.Errorf("failed to marshal metadata: %w", err) return 0, fmt.Errorf("failed to marshal metadata: %w", err)
} }
// Metadata write failure is non-fatal: the content is then served as // Metadata write failure is non-fatal; content is already stored.
// application/octet-stream. err = os.WriteFile(metaPath, metaData, StorageFilePerm)
err = writeFileSynced(metaPath, metaData)
if err != nil { if err != nil {
s.log.Warn("failed to write variant metadata sidecar", s.log.Warn("failed to write variant metadata sidecar",
"path", metaPath, "error", err) "path", metaPath, "error", err)
} }
err = writeFileSynced(path, data)
if err != nil {
return 0, err
}
return size, nil return size, nil
} }
@@ -185,6 +185,7 @@ func SetupTestService(
cache, err := NewCache(db, CacheConfig{ cache, err := NewCache(db, CacheConfig{
StateDir: tmpDir, StateDir: tmpDir,
CacheTTL: time.Hour,
NegativeTTL: 5 * time.Minute, NegativeTTL: 5 * time.Minute,
}) })
if err != nil { if err != nil {
+2 -5
View File
@@ -7,8 +7,6 @@ import (
"regexp" "regexp"
"strconv" "strconv"
"strings" "strings"
"sneak.berlin/go/pixa/internal/imageprocessor"
) )
// URL parsing errors. // URL parsing errors.
@@ -22,9 +20,8 @@ var (
ErrPathTraversal = errors.New("path traversal detected") ErrPathTraversal = errors.New("path traversal detected")
) )
// MaxDimension is the maximum allowed width or height, the same as the image // MaxDimension is the maximum allowed width or height.
// processor's maximum for an output image. const MaxDimension = 8192
const MaxDimension = imageprocessor.MaxOutputDimension
// ValidateDimension checks one requested width or height; name ("width" or // ValidateDimension checks one requested width or height; name ("width" or
// "height") appears in the error. 0 means "original size" and is valid. // "height") appears in the error. 0 means "original size" and is valid.
+3 -5
View File
@@ -98,10 +98,9 @@ func (s *Middleware) ClientIP() func(http.Handler) http.Handler {
// holds whole; an IPv4-mapped address (::ffff:a.b.c.d) is counted as the // holds whole; an IPv4-mapped address (::ffff:a.b.c.d) is counted as the
// IPv4 address it carries, since every such address falls in the same /64. // IPv4 address it carries, since every such address falls in the same /64.
// Counts are kept only for the current and the previous window, so memory // Counts are kept only for the current and the previous window, so memory
// stays bounded. limitHandler writes the 429 after Retry-After is set; nil // stays bounded.
// keeps the rate limit library's plain-text "Too Many Requests".
func (s *Middleware) RateLimit( func (s *Middleware) RateLimit(
requestLimit int, window time.Duration, limitHandler http.HandlerFunc, requestLimit int, window time.Duration,
) func(http.Handler) http.Handler { ) func(http.Handler) http.Handler {
return httprate.LimitBy(requestLimit, window, return httprate.LimitBy(requestLimit, window,
func(r *http.Request) (string, error) { func(r *http.Request) (string, error) {
@@ -113,8 +112,7 @@ func (s *Middleware) RateLimit(
} }
return httprate.CanonicalizeIP(ip), nil return httprate.CanonicalizeIP(ip), nil
}, })
httprate.WithLimitHandler(limitHandler))
} }
// requestIDPattern is what a request's own X-Request-Id must look like to be // requestIDPattern is what a request's own X-Request-Id must look like to be
@@ -1,153 +0,0 @@
package server
import (
"bytes"
"encoding/json"
"image/jpeg"
"net/http"
"net/http/httptest"
"testing"
"sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imgcache"
)
// TestImageProxyRefusesComputedSideAboveMaximum requests sizes of a 3000x600
// and a 600x3000 PNG through both image routes. A size with one side 0, whose
// side computed from the source's proportions would be above 8192, is refused
// with 400 and the JSON error body. The sizes after those refusals are served
// as before, which also shows that a refusal is not cached as a failed fetch of
// the source: orig, a size with both sides given, and a size with one side 0
// whose computed side stays within 8192.
func TestImageProxyRefusesComputedSideAboveMaximum(t *testing.T) {
t.Parallel()
// The upstream paths of the two sources
const (
widePath = "/wide.png"
tallPath = "/tall.png"
)
wide := encodeTestPNG(t, 3000, 600)
tall := encodeTestPNG(t, 600, 3000)
upstream := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "image/png")
if r.URL.Path == tallPath {
_, _ = w.Write(tall)
return
}
_, _ = w.Write(wide)
}))
t.Cleanup(upstream.Close)
s, _, _ := startImageProxy(t, upstream)
generator, err := encurl.NewGenerator(testSigningKey)
if err != nil {
t.Fatalf("encurl.NewGenerator: %v", err)
}
tests := []struct {
path string
// size is the size of the /v1/image/ URL, and width and height are
// those of the encrypted URL
size string
width, height int
// wantWidth and wantHeight are the served image's, 0 for a refusal
wantWidth, wantHeight int
}{
{widePath, "0x8192", 0, 8192, 0, 0},
{tallPath, "8192x0", 8192, 0, 0, 0},
{widePath, "orig", 0, 0, 3000, 600},
{widePath, "300x200", 300, 200, 300, 200},
{widePath, "0x300", 0, 300, 1500, 300},
{tallPath, "300x0", 300, 0, 300, 1500},
}
for _, tt := range tests {
token, err := generator.Generate(&encurl.Payload{
SourceHost: upstreamHost, SourcePath: tt.path,
Width: tt.width, Height: tt.height, Format: imgcache.FormatJPEG,
})
if err != nil {
t.Fatalf("encurl Generate: %v", err)
}
for _, target := range []string{
"/v1/image/" + upstreamHost + tt.path + "/" + tt.size + ".jpeg",
"/v1/e/" + token + "/img.jpg",
} {
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d, Content-Type %s",
target, rec.Code, rec.Header().Get("Content-Type"))
if tt.wantWidth == 0 {
checkSizeRefused(t, rec)
} else {
checkServedSize(t, rec, tt.wantWidth, tt.wantHeight)
}
}
}
}
// checkSizeRefused checks that rec answers 400 with the JSON error body that
// refuses a size whose side computed from the source's proportions is above
// 8192.
func checkSizeRefused(t *testing.T, rec *httptest.ResponseRecorder) {
t.Helper()
const wantError = "size too large for the source: " +
"the side computed from its proportions is above 8192"
var body struct {
Error string `json:"error"`
Status int `json:"status"`
}
err := json.Unmarshal(rec.Body.Bytes(), &body)
if err != nil {
t.Errorf("status %d, body is not JSON: %v", rec.Code, err)
return
}
if rec.Code != http.StatusBadRequest || body.Status != http.StatusBadRequest ||
body.Error != wantError {
t.Errorf("status %d, body %+v, want %d and the error %q",
rec.Code, body, http.StatusBadRequest, wantError)
}
}
// checkServedSize checks that rec answers 200 with a JPEG of wantWidth by
// wantHeight.
func checkServedSize(
t *testing.T, rec *httptest.ResponseRecorder, wantWidth, wantHeight int,
) {
t.Helper()
if rec.Code != http.StatusOK {
t.Errorf("status %d, want %d; body %s",
rec.Code, http.StatusOK, rec.Body.String())
return
}
served, err := jpeg.DecodeConfig(bytes.NewReader(rec.Body.Bytes()))
if err != nil {
t.Errorf("decoding the served JPEG: %v", err)
return
}
if served.Width != wantWidth || served.Height != wantHeight {
t.Errorf("served %dx%d, want %dx%d",
served.Width, served.Height, wantWidth, wantHeight)
}
}
@@ -1,91 +0,0 @@
package server
import (
"fmt"
"net/http"
"net/http/httptest"
"testing"
"sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imgcache"
)
// TestImageProxyRefusesFitOutsideAboveMaximum requests sizes with fit=outside
// through both image routes. fit=outside scales the source to cover the size,
// and so past it on one side: a size for which that side would be above 8192
// is refused with 400 and the JSON error body, whether it is the width or the
// height, and a size for which it stays within 8192 is served at the scaled
// size.
func TestImageProxyRefusesFitOutsideAboveMaximum(t *testing.T) {
t.Parallel()
// The upstream paths of the sources
const (
widePath = "/wide.png"
rowPath = "/row.png"
columnPath = "/column.png"
)
sources := map[string][]byte{
widePath: encodeTestPNG(t, 3000, 600),
rowPath: encodeTestPNG(t, 8192, 1),
columnPath: encodeTestPNG(t, 1, 8192),
}
upstream := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "image/png")
_, _ = w.Write(sources[r.URL.Path])
}))
t.Cleanup(upstream.Close)
s, _, _ := startImageProxy(t, upstream)
generator, err := encurl.NewGenerator(testSigningKey)
if err != nil {
t.Fatalf("encurl.NewGenerator: %v", err)
}
tests := []struct {
path string
width, height int
// wantWidth and wantHeight are the served image's, 0 for a refusal
wantWidth, wantHeight int
}{
{widePath, 1, 8192, 0, 0},
{rowPath, 1, 8192, 0, 0},
{columnPath, 8192, 1, 0, 0},
{widePath, 1, 1200, 6000, 1200},
}
for _, tt := range tests {
token, err := generator.Generate(&encurl.Payload{
SourceHost: upstreamHost, SourcePath: tt.path,
Width: tt.width, Height: tt.height, Format: imgcache.FormatJPEG,
FitMode: imgcache.FitOutside,
})
if err != nil {
t.Fatalf("encurl Generate: %v", err)
}
size := fmt.Sprintf("%dx%d", tt.width, tt.height)
for _, target := range []string{
"/v1/image/" + upstreamHost + tt.path + "/" + size +
".jpeg?fit=outside",
"/v1/e/" + token + "/img.jpg",
} {
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d, Content-Type %s",
target, rec.Code, rec.Header().Get("Content-Type"))
if tt.wantWidth == 0 {
checkSizeRefused(t, rec)
} else {
checkServedSize(t, rec, tt.wantWidth, tt.wantHeight)
}
}
}
}
@@ -1,204 +0,0 @@
package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"testing"
"sneak.berlin/go/pixa/internal/config"
)
// requestImageFrom sends method path to s from remoteAddr and returns the
// response. The image handlers refuse an unsigned image URL with 401 and a
// token they cannot decrypt with 400, so either status shows the limit let
// the request through.
func requestImageFrom(
t *testing.T, s *Server, method, path, remoteAddr string,
) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequestWithContext(t.Context(), method, path, nil)
req.RemoteAddr = remoteAddr
rec := httptest.NewRecorder()
s.ServeHTTP(rec, req)
return rec
}
// TestImageRequestLimitRefusesRequestOverLimit verifies that GET and HEAD on
// both image routes count toward one limit per client, and that the request
// after image_requests_per_minute of them is refused with 429 and a
// Retry-After header.
func TestImageRequestLimitRefusesRequestOverLimit(t *testing.T) {
t.Parallel()
requests := []struct {
method string
path string
want int
}{
{http.MethodGet, unsignedImagePath, http.StatusUnauthorized},
{http.MethodHead, unsignedImagePath, http.StatusUnauthorized},
{http.MethodGet, encryptedImagePath, http.StatusBadRequest},
}
s := newTestServer(t)
s.config.ImageRequestsPerMinute = len(requests)
s.SetupRoutes()
for _, tc := range requests {
rec := requestImageFrom(t, s, tc.method, tc.path, firstClient)
if rec.Code != tc.want {
t.Fatalf("%s %s within the limit status = %d, want %d",
tc.method, tc.path, rec.Code, tc.want)
}
}
rec := requestImageFrom(t, s, http.MethodHead, encryptedImagePath, firstClient)
if rec.Code != http.StatusTooManyRequests {
t.Fatalf("request over the limit status = %d, want %d",
rec.Code, http.StatusTooManyRequests)
}
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)
}
}
// TestImageRequestLimitAnswersWithJSONError verifies that the 429 for a
// request over the limit carries Retry-After and the JSON error body the
// image routes send for their other errors.
func TestImageRequestLimitAnswersWithJSONError(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.ImageRequestsPerMinute = 1
s.SetupRoutes()
requestImageFrom(t, s, http.MethodGet, unsignedImagePath, firstClient)
rec := requestImageFrom(t, s, http.MethodGet, unsignedImagePath, firstClient)
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
if rec.Code != http.StatusTooManyRequests {
t.Fatalf("request over the limit status = %d, want %d",
rec.Code, http.StatusTooManyRequests)
}
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)
}
contentType := rec.Header().Get("Content-Type")
if contentType != "application/json" {
t.Errorf("Content-Type = %q, want %q", contentType, "application/json")
}
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.StatusTooManyRequests ||
body.Timestamp == "" {
t.Errorf("body = %+v, want an error, status %d and a timestamp",
body, http.StatusTooManyRequests)
}
}
// TestImageRequestLimitCountsClientsSeparately verifies that one client going
// over the limit does not limit another.
func TestImageRequestLimitCountsClientsSeparately(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.ImageRequestsPerMinute = 1
s.SetupRoutes()
rec := requestImageFrom(t, s, http.MethodGet, unsignedImagePath, firstClient)
if rec.Code != http.StatusUnauthorized {
t.Fatalf("first client's first request status = %d, want %d",
rec.Code, http.StatusUnauthorized)
}
rec = requestImageFrom(t, s, http.MethodGet, unsignedImagePath, firstClient)
if rec.Code != http.StatusTooManyRequests {
t.Fatalf("first client's second request status = %d, want %d",
rec.Code, http.StatusTooManyRequests)
}
rec = requestImageFrom(t, s, http.MethodGet, unsignedImagePath, secondClient)
if rec.Code != http.StatusUnauthorized {
t.Errorf("second client's first request status = %d, want %d",
rec.Code, http.StatusUnauthorized)
}
}
// TestImageRequestLimitZeroLetsEveryRequestThrough verifies that with
// image_requests_per_minute set to 0 no image request is refused with 429,
// not even past the default limit.
func TestImageRequestLimitZeroLetsEveryRequestThrough(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.ImageRequestsPerMinute = 0
s.SetupRoutes()
for request := range config.DefaultImageRequestsPerMinute + 1 {
rec := requestImageFrom(t, s, http.MethodGet, unsignedImagePath, firstClient)
if rec.Code != http.StatusUnauthorized {
t.Fatalf("request %d status = %d, want %d",
request+1, rec.Code, http.StatusUnauthorized)
}
}
}
// TestImageRequestLimitDoesNotCountPreflight verifies that a browser's CORS
// preflight request to an image route is answered without being counted, so
// a client that has sent several still gets its image request through.
func TestImageRequestLimitDoesNotCountPreflight(t *testing.T) {
t.Parallel()
s := newTestServer(t)
s.config.AccessControlAllowOrigin = config.DefaultAccessControlAllowOrigin
s.config.ImageRequestsPerMinute = 1
s.SetupRoutes()
for range 3 {
req := httptest.NewRequestWithContext(
t.Context(), http.MethodOptions, unsignedImagePath, nil)
req.RemoteAddr = firstClient
req.Header.Set("Origin", "https://app.example.com")
req.Header.Set("Access-Control-Request-Method", http.MethodGet)
rec := httptest.NewRecorder()
s.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("preflight status = %d, want %d", rec.Code, http.StatusOK)
}
}
rec := requestImageFrom(t, s, http.MethodGet, unsignedImagePath, firstClient)
if rec.Code != http.StatusUnauthorized {
t.Errorf("image request after preflights status = %d, want %d",
rec.Code, http.StatusUnauthorized)
}
}
+1 -10
View File
@@ -66,7 +66,7 @@ func (s *Server) SetupRoutes() {
r.Use(s.h.LimitBody(handlers.MaxFormBytes)) r.Use(s.h.LimitBody(handlers.MaxFormBytes))
r.Use(s.h.CSRF()) r.Use(s.h.CSRF())
r.Get("/", s.h.HandleRoot()) r.Get("/", s.h.HandleRoot())
r.With(s.mw.RateLimit(LoginAttemptsPerMinute, time.Minute, nil)). r.With(s.mw.RateLimit(LoginAttemptsPerMinute, time.Minute)).
Post("/", s.h.HandleRoot()) Post("/", s.h.HandleRoot())
r.Post("/generate", s.h.HandleGenerateURL()) r.Post("/generate", s.h.HandleGenerateURL())
}) })
@@ -88,15 +88,6 @@ func (s *Server) SetupRoutes() {
r.Group(func(r chi.Router) { r.Group(func(r chi.Router) {
r.Use(s.refuseDuringMaintenance) r.Use(s.refuseDuringMaintenance)
// Limited per client, as the login POST is, but the 429 has
// the image routes' JSON error body; 0 turns the limit off.
// The CORS middleware answers a preflight before it gets
// here, so a preflight is not counted.
if s.config.ImageRequestsPerMinute > 0 {
r.Use(s.mw.RateLimit(s.config.ImageRequestsPerMinute,
time.Minute, s.h.HandleTooManyRequests()))
}
// Main image proxy route // Main image proxy route
// /v1/image/<host>/<path>/<width>x<height>.<format>, or with no // /v1/image/<host>/<path>/<width>x<height>.<format>, or with no
// format /v1/image/<host>/<path>/<width>x<height> // format /v1/image/<host>/<path>/<width>x<height>
@@ -1,71 +0,0 @@
package server
import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"sneak.berlin/go/pixa/internal/encurl"
"sneak.berlin/go/pixa/internal/imgcache"
)
// svgServedAsAVIF is an SVG whose first bytes pass the first-bytes check for
// AVIF: its XML comment puts "ftypavis" at bytes 4 to 11, where an AVIF file
// has its ftyp box and brand. libvips would load it as SVG.
const svgServedAsAVIF = `<!--ftypavis--><svg xmlns="http://www.w3.org/2000/svg" ` +
`width="64" height="48"><rect width="64" height="48"/></svg>`
// TestImageProxyRefusesSVGServedAsAVIF requests svgServedAsAVIF, which the
// upstream answers as image/avif, through both image routes, resized as PNG
// and in its own format (orig). Each request is answered 415, as for an SVG
// source, never with an image.
func TestImageProxyRefusesSVGServedAsAVIF(t *testing.T) {
t.Parallel()
upstream := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "image/avif")
_, _ = io.WriteString(w, svgServedAsAVIF)
}))
t.Cleanup(upstream.Close)
s, _, _ := startImageProxy(t, upstream)
generator, err := encurl.NewGenerator(testSigningKey)
if err != nil {
t.Fatalf("encurl.NewGenerator: %v", err)
}
encryptedURL := func(format imgcache.ImageFormat) string {
token, err := generator.Generate(&encurl.Payload{
SourceHost: upstreamHost, SourcePath: "/drawing.avif",
Width: 32, Height: 24, Format: format,
})
if err != nil {
t.Fatalf("encurl Generate: %v", err)
}
return "/v1/e/" + token + "/drawing." + string(format)
}
for _, target := range []string{
"/v1/image/" + upstreamHost + "/drawing.avif/32x24.png",
"/v1/image/" + upstreamHost + "/drawing.avif/64x48.orig",
encryptedURL(imgcache.FormatPNG),
encryptedURL(imgcache.FormatOriginal),
} {
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d, Content-Type %s, body %q",
target, rec.Code, rec.Header().Get("Content-Type"), rec.Body.String())
if rec.Code != http.StatusUnsupportedMediaType ||
!strings.Contains(rec.Body.String(), "unsupported source format") {
t.Errorf("GET %s: %d %q, want %d unsupported source format",
target, rec.Code, rec.Body.String(), http.StatusUnsupportedMediaType)
}
}
}
-41
View File
@@ -1,41 +0,0 @@
package server
import (
"io"
"net/http"
"net/http/httptest"
"testing"
)
// TestImageProxyRefusesSVG requests an image the upstream answers as SVG,
// in its own format (orig) and resized as PNG, through pixa's router,
// upstream fetcher and image processor. pixa does not accept SVG as a
// source, so each request is answered 415, never with an image.
func TestImageProxyRefusesSVG(t *testing.T) {
t.Parallel()
upstream := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "image/svg+xml")
_, _ = io.WriteString(w, `<svg xmlns="http://www.w3.org/2000/svg" `+
`width="64" height="48"><rect width="64" height="48"/></svg>`)
}))
t.Cleanup(upstream.Close)
s, _, _ := startImageProxy(t, upstream)
for _, sizeFormat := range []string{"64x48.orig", "32x24.png"} {
target := "/v1/image/" + upstreamHost + "/drawing.svg/" + sizeFormat
rec := httptest.NewRecorder()
s.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, target, nil))
t.Logf("GET %s: %d, Content-Type %s, body %s",
target, rec.Code, rec.Header().Get("Content-Type"), rec.Body.String())
if rec.Code != http.StatusUnsupportedMediaType {
t.Errorf("GET %s status = %d, want %d",
target, rec.Code, http.StatusUnsupportedMediaType)
}
}
}
+1 -3
View File
@@ -94,8 +94,7 @@ main() {
} }
# start_containers starts a new origin and a new pixad, and waits up to 30 # start_containers starts a new origin and a new pixad, and waits up to 30
# seconds for pixad's health check to pass. pixad sees all of vegeta's # seconds for pixad's health check to pass.
# requests come from one address, so its image request limit is off.
start_containers() { start_containers() {
docker run -d --name "$run-origin" \ docker run -d --name "$run-origin" \
--network "$run" --network-alias origin \ --network "$run" --network-alias origin \
@@ -105,7 +104,6 @@ start_containers() {
--network "$run" -p 127.0.0.1::8080 --health-interval=1s \ --network "$run" -p 127.0.0.1::8080 --health-interval=1s \
-e PIXA_SIGNING_KEY="$(head -c 32 /dev/urandom | base64)" \ -e PIXA_SIGNING_KEY="$(head -c 32 /dev/urandom | base64)" \
-e PIXA_ALLOWLIST_HOSTS=origin -e PIXA_ALLOW_HTTP=true \ -e PIXA_ALLOWLIST_HOSTS=origin -e PIXA_ALLOW_HTTP=true \
-e PIXA_IMAGE_REQUESTS_PER_MINUTE=0 \
"$image" >/dev/null "$image" >/dev/null
waited=0 waited=0