Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
afb5dc5d0d | ||
|
|
d8760c40a1 | ||
|
|
ca512fdb4f | ||
|
|
52f69d3082 | ||
|
|
162a2a33ef | ||
|
|
40ad39843b |
@@ -46,9 +46,9 @@ everything in this list without further settings. The reverse proxy must:
|
||||
builds encrypted URLs from `Host`, and checks `Referer` against
|
||||
`referer_blocklist`;
|
||||
- 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
|
||||
counts each client by its own address (see `trusted_proxies` under
|
||||
Configuration);
|
||||
the address pixa sees the proxy's requests come from, so the login limit and
|
||||
`image_requests_per_minute` count each client by its own address (see
|
||||
`trusted_proxies` under Configuration);
|
||||
- wait for pixa's answer for at least `downstream_timeout` (default `60s`), the
|
||||
longest pixa takes to fetch, convert and send an image.
|
||||
|
||||
@@ -58,8 +58,9 @@ password there.
|
||||
|
||||
pixa does the rest itself: it checks signatures and encrypted URLs, applies the
|
||||
allowlist, refuses upstream hosts with private or local addresses, limits login
|
||||
attempts, upstream response size and image dimensions, and sends the security
|
||||
headers, `Strict-Transport-Security` included, with every response.
|
||||
attempts, each client's image requests, upstream response size and image
|
||||
dimensions, and sends the security headers, `Strict-Transport-Security`
|
||||
included, with every response.
|
||||
|
||||
The state directory (`state_dir`, `/var/lib/pixa` in the container) holds the
|
||||
database and the disk cache:
|
||||
@@ -83,12 +84,12 @@ which answers 200 whenever pixa is running, in maintenance mode too (see
|
||||
`maintenance_mode`).
|
||||
|
||||
On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in
|
||||
progress, the images being processed and the counts of cache hits, misses,
|
||||
fetches and conversions being written to the database 5 seconds to finish, and
|
||||
exits: with 0, or with 1 when images were still being processed or counts still
|
||||
being written after those 5 seconds, 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.
|
||||
progress and the images being processed 5 seconds to finish, then writes to the
|
||||
database the counts of cache hits, misses, fetches and conversions that requests
|
||||
could not write by their deadline, and exits: with 0, or with 1 when images were
|
||||
still being processed after those 5 seconds, some of those counts could not be
|
||||
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
|
||||
it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL
|
||||
@@ -188,7 +189,10 @@ path under `/v1/` answers 200, in maintenance mode too.
|
||||
`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
|
||||
redirects to, is `localhost`, ends in `.localhost` or `.local`, or has an
|
||||
address in a blocked network (see `blocked_networks`); 502 when the upstream
|
||||
address in a blocked network (see `blocked_networks`); 429 over
|
||||
`image_requests_per_minute` (below); 415 when the upstream answers with a
|
||||
`Content-Type` that is not one of the source formats (below), or with an image
|
||||
whose first bytes libvips reads as another format; 502 when the upstream
|
||||
answered with an error status, and for 5 minutes after that for the same
|
||||
source URL; 503 when pixa is busy or in maintenance mode; 500 for any other
|
||||
failure.
|
||||
@@ -199,7 +203,7 @@ path under `/v1/` answers 200, in maintenance mode too.
|
||||
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
|
||||
itself is still arriving; 400 for an `Accept` header that is not valid, and
|
||||
406, 403, 502, 503 and 500, as for `/v1/image/`.
|
||||
406, 403, 415, 429, 502, 503 and 500, as for `/v1/image/`.
|
||||
- `GET /robots.txt` — asks every crawler to stay away (`Disallow: /`). Needs:
|
||||
nothing. Answers: 200.
|
||||
- `GET /.well-known/healthcheck.json` — JSON with `status` (`ok`), `now`,
|
||||
@@ -259,9 +263,17 @@ once, is refused with 400.
|
||||
both optional (values under Signature Specification). Both are part of what is
|
||||
cached, so each value of either is a separate cached image.
|
||||
|
||||
The source image may be JPEG, PNG, GIF, WebP, AVIF, JPEG XL or SVG. The
|
||||
upstream's `Content-Type` must name its format, and the image's first bytes must
|
||||
match it.
|
||||
A size with one side `0` has that side scaled to keep the source's proportions,
|
||||
and is refused with 400 when that side would come out above 8192, the most
|
||||
either side may be; so is a size with `fit=outside`, which scales the image to
|
||||
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
|
||||
`Accept` header, in this order:
|
||||
@@ -315,6 +327,15 @@ its own `X-Forwarded-For`, whether it connects directly or through the proxy,
|
||||
because its own address is trusted too. Setting `trusted_proxies` to only the
|
||||
address pixa sees for requests that come through the proxy closes this.
|
||||
|
||||
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
|
||||
|
||||
An encrypted URL is an image URL made on pixa's own web page by someone who
|
||||
@@ -330,7 +351,8 @@ nor change what it asks for.
|
||||
(`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
|
||||
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.
|
||||
or `0`, that side is scaled to keep the image's proportions. A URL whose
|
||||
scaled side would come out above 8192 answers 400.
|
||||
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
|
||||
with `http` instead while `debug` is on. The name after the token is ignored
|
||||
@@ -484,6 +506,7 @@ 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` | `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_IMAGE_REQUESTS_PER_MINUTE` | `image_requests_per_minute` | Image requests per client per minute; `0` turns it off; default `600` |
|
||||
| `PIXA_UPSTREAM_FETCH_TIMEOUT` | `upstream_fetch_timeout` | Time allowed for one fetch from an upstream host; default `30s` |
|
||||
| `PIXA_UPSTREAM_MAX_RESPONSE_SIZE` | `upstream_max_response_size` | Largest upstream response accepted, in bytes; default 50 MiB |
|
||||
| `PIXA_DOWNSTREAM_TIMEOUT` | `downstream_timeout` | Time allowed for answering one client request; default `60s` |
|
||||
@@ -533,13 +556,13 @@ Key settings in more detail:
|
||||
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
|
||||
`172.17.0.1/32`, when the defaults do not cover it, or to trust nothing else
|
||||
(see the login limit under Routes). For a proxy on the Docker host that
|
||||
connects to pixa over `127.0.0.1`, that address is the gateway of the
|
||||
container's Docker network (`172.17.0.1` on the default bridge), not the
|
||||
proxy's own address; a proxy that connects through another of the host's
|
||||
addresses is seen with that address. To be sure which address it is, set this
|
||||
to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request through the
|
||||
proxy, and read `remoteIP` in pixa's request log line for it
|
||||
(see the login limit and `image_requests_per_minute` under Routes). For a
|
||||
proxy on the Docker host that connects to pixa over `127.0.0.1`, that address
|
||||
is the gateway of the container's Docker network (`172.17.0.1` on the default
|
||||
bridge), not the proxy's own address; a proxy that connects through another of
|
||||
the host's addresses is seen with that address. To be sure which address it
|
||||
is, set this to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request
|
||||
through the proxy, and read `remoteIP` in pixa's request log line for it
|
||||
- `upstream_fetch_timeout` — time allowed for one fetch from an upstream host,
|
||||
as a duration such as `30s` (the default) or `2m`
|
||||
- `upstream_max_response_size` — largest upstream response accepted, in bytes;
|
||||
@@ -572,6 +595,12 @@ Key settings in more detail:
|
||||
container's CPU limit. A request that finds all of them in use waits up to 10
|
||||
seconds for one to free up; if none does, and `downstream_timeout` has not
|
||||
ended first, it is answered 503 the same way
|
||||
- `image_requests_per_minute` — how many requests one client may send to the
|
||||
image routes per minute (see the limit on them under Routes); default `600`.
|
||||
`0` turns the limit off, and a negative or non-integer value aborts startup.
|
||||
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
|
||||
`/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`)
|
||||
|
||||
@@ -26,25 +26,68 @@ resolved. The disk cache is now size-bounded with LRU eviction
|
||||
|
||||
# Next Step
|
||||
|
||||
P2: security: per-IP rate limiting on the image routes
|
||||
P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-08 a request past its deadline no longer waits for a database write
|
||||
(closes #224). On a busy host, `TestService_Get_ReturnsByItsDeadline` failed
|
||||
because its request returned over a second after its deadline: it was still
|
||||
waiting for its miss count to be written to the database, a write with no
|
||||
deadline, which in pixad waits for the one database connection that every
|
||||
request shares. The hit, miss, upstream fetch and transform counts are now
|
||||
each written in a goroutine of their own, which the request waits for only
|
||||
until its deadline; a count not written by then is written after the request
|
||||
returns. On shutdown pixad waits for those writes within the same 5 seconds as
|
||||
for the images being processed, and exits with 1 when some are unfinished. 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-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
|
||||
(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
|
||||
@@ -751,8 +794,6 @@ P2: security: per-IP rate limiting on the image routes
|
||||
|
||||
# Future Steps
|
||||
|
||||
- P2: security
|
||||
- per-origin rate limiting
|
||||
- P2: HTTP response handling
|
||||
- Last-Modified headers
|
||||
- P2: configuration
|
||||
|
||||
@@ -115,6 +115,14 @@ upstream_connections: 64
|
||||
# frees up it is answered 503, unless downstream_timeout has ended first.
|
||||
# max_concurrent_processing: 4
|
||||
|
||||
# How many requests to the image routes (/v1/image/ and /v1/e/) one client
|
||||
# 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)
|
||||
upstream_fetch_timeout: 30s
|
||||
|
||||
|
||||
+39
-10
@@ -34,6 +34,7 @@ const (
|
||||
DefaultUpstreamFetchTimeout = 30 * time.Second
|
||||
DefaultUpstreamMaxResponseSize = 50 << 20 // 50 MiB
|
||||
DefaultDownstreamTimeout = 60 * time.Second
|
||||
DefaultImageRequestsPerMinute = 600
|
||||
)
|
||||
|
||||
// Configuration key names.
|
||||
@@ -61,6 +62,7 @@ const (
|
||||
keyUpstreamFetchTimeout = "upstream_fetch_timeout"
|
||||
keyUpstreamMaxResponseSize = "upstream_max_response_size"
|
||||
keyDownstreamTimeout = "downstream_timeout"
|
||||
keyImageRequestsPerMinute = "image_requests_per_minute"
|
||||
)
|
||||
|
||||
// placeholderSigningKey is the dummy signing_key shipped in
|
||||
@@ -158,6 +160,11 @@ type Config struct {
|
||||
// of the per-request timeout middleware.
|
||||
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
|
||||
// addition to the built-in SSRF blocklist. Enforced by the upstream
|
||||
// fetcher's dialer; the built-in ranges always apply.
|
||||
@@ -302,6 +309,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
keyAccessControlAllowOrigin, DefaultAccessControlAllowOrigin),
|
||||
DownstreamTimeout: loader.durationVal(
|
||||
keyDownstreamTimeout, DefaultDownstreamTimeout),
|
||||
ImageRequestsPerMinute: loader.intVal(
|
||||
keyImageRequestsPerMinute, DefaultImageRequestsPerMinute),
|
||||
CacheMaxBytes: loader.int64Val(keyCacheMaxBytes, 0),
|
||||
BlockedNetworks: blockedNetworks,
|
||||
TrustedProxies: trustedProxies,
|
||||
@@ -311,9 +320,7 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
// The default for an omitted cache_max_bytes is worked out when
|
||||
// the cache opens; here we only record whether the operator set
|
||||
// the key explicitly.
|
||||
if _, present := lookupValue(sc, keyCacheMaxBytes); present {
|
||||
c.CacheMaxBytesExplicit = true
|
||||
}
|
||||
_, c.CacheMaxBytesExplicit = lookupValue(sc, keyCacheMaxBytes)
|
||||
|
||||
// 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
|
||||
@@ -426,7 +433,7 @@ func isKnownConfigKey(key string) bool {
|
||||
keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks,
|
||||
keyTrustedProxies, keyAccessControlAllowOrigin, keyUpstreamFetchTimeout,
|
||||
keyUpstreamMaxResponseSize, keyDownstreamTimeout, keyRefererBlocklist,
|
||||
"env":
|
||||
keyImageRequestsPerMinute, "env":
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -461,6 +468,7 @@ func envVarNames() map[string]string {
|
||||
keyUpstreamFetchTimeout: "PIXA_UPSTREAM_FETCH_TIMEOUT",
|
||||
keyUpstreamMaxResponseSize: "PIXA_UPSTREAM_MAX_RESPONSE_SIZE",
|
||||
keyDownstreamTimeout: "PIXA_DOWNSTREAM_TIMEOUT",
|
||||
keyImageRequestsPerMinute: "PIXA_IMAGE_REQUESTS_PER_MINUTE",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -613,6 +621,14 @@ func (c *Config) validate() error {
|
||||
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()
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -625,12 +641,9 @@ func (c *Config) validate() error {
|
||||
}
|
||||
}
|
||||
|
||||
if c.SentryDSN != "" {
|
||||
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)
|
||||
}
|
||||
err = c.validateSentryDSN()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if (c.MetricsUsername == "") != (c.MetricsPassword == "") {
|
||||
@@ -642,6 +655,22 @@ func (c *Config) validate() error {
|
||||
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
|
||||
// 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
|
||||
|
||||
@@ -70,6 +70,7 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
|
||||
t.Setenv("PIXA_UPSTREAM_CONNECTIONS_PER_HOST", "5")
|
||||
t.Setenv("PIXA_UPSTREAM_CONNECTIONS", "10")
|
||||
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_BLOCKED_NETWORKS", "203.0.113.0/24")
|
||||
t.Setenv("PIXA_TRUSTED_PROXIES", "192.0.2.0/24")
|
||||
@@ -99,6 +100,7 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
|
||||
UpstreamConnectionsPerHost: 5,
|
||||
UpstreamConnections: 10,
|
||||
MaxConcurrentProcessing: 3,
|
||||
ImageRequestsPerMinute: 30,
|
||||
CacheMaxBytes: 1024,
|
||||
CacheMaxBytesExplicit: true,
|
||||
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")},
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
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)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,7 @@ package handlers
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"time"
|
||||
@@ -71,16 +72,21 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
}
|
||||
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // the eviction loop outlives OnStart; OnStop cancels it
|
||||
//nolint:contextcheck // the cache's goroutines outlive OnStart; OnStop stops them
|
||||
OnStart: func(_ context.Context) error {
|
||||
return s.initImageService()
|
||||
},
|
||||
// The pending counts are written here, before the database's stop
|
||||
// hook closes it.
|
||||
OnStop: func(ctx context.Context) error {
|
||||
if s.imgCache == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return s.imgCache.StopEviction(ctx)
|
||||
return errors.Join(
|
||||
s.imgCache.StopEviction(ctx),
|
||||
s.imgCache.StopPendingCountWrites(ctx),
|
||||
)
|
||||
},
|
||||
})
|
||||
|
||||
@@ -93,13 +99,6 @@ func (s *Handlers) WaitForProcessing(ctx context.Context) int {
|
||||
return s.imgSvc.WaitForProcessing(ctx)
|
||||
}
|
||||
|
||||
// WaitForCountWrites waits until every count a request has started writing
|
||||
// to the database is written, or until ctx ends, and reports whether they
|
||||
// all were.
|
||||
func (s *Handlers) WaitForCountWrites(ctx context.Context) bool {
|
||||
return s.imgSvc.WaitForCountWrites(ctx)
|
||||
}
|
||||
|
||||
// newCacheConfig builds the image cache's configuration from cfg.
|
||||
// cache_max_bytes: 0 disables the disk cache entirely; any other value
|
||||
// is the eviction limit in bytes; when it is omitted, the cache works
|
||||
@@ -107,7 +106,6 @@ func (s *Handlers) WaitForCountWrites(ctx context.Context) bool {
|
||||
func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig {
|
||||
return imgcache.CacheConfig{
|
||||
StateDir: cfg.StateDir,
|
||||
CacheTTL: imgcache.DefaultCacheTTL,
|
||||
NegativeTTL: imgcache.DefaultNegativeTTL,
|
||||
MaxBytes: cfg.CacheMaxBytes,
|
||||
UseDefaultMaxBytes: !cfg.CacheMaxBytesExplicit,
|
||||
@@ -129,6 +127,9 @@ func (s *Handlers) initImageService() error {
|
||||
// write-pressure passes. No-op when the disk cache is disabled.
|
||||
cache.StartEviction(imgcache.DefaultEvictionInterval)
|
||||
|
||||
// Writes the counts requests could not write by their deadline
|
||||
cache.StartPendingCountWrites()
|
||||
|
||||
// Create the fetcher config
|
||||
fetcherCfg := httpfetcher.DefaultConfig()
|
||||
fetcherCfg.AllowHTTP = s.config.AllowHTTP
|
||||
|
||||
@@ -47,7 +47,6 @@ func setupTestHandler(t *testing.T) *testFixtures {
|
||||
|
||||
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -206,6 +206,16 @@ func parseExpires(query url.Values) (time.Time, error) {
|
||||
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.
|
||||
func (s *Handlers) respondImageError(
|
||||
w http.ResponseWriter, req *imgcache.ImageRequest, err error,
|
||||
@@ -237,6 +247,22 @@ func (s *Handlers) respondImageError(
|
||||
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)
|
||||
}
|
||||
|
||||
|
||||
@@ -31,7 +31,6 @@ func newSignedHostServer(
|
||||
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -29,7 +29,6 @@ func newImageRoute(t *testing.T, fetcher httpfetcher.Fetcher) http.Handler {
|
||||
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -161,6 +161,14 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
|
||||
errors.Is(err, imageprocessor.ErrTooManyImages):
|
||||
s.respondError(w, "server busy, try again later",
|
||||
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:
|
||||
s.log.Error("image request failed", "error", err)
|
||||
s.respondError(w, "internal error", http.StatusInternalServerError)
|
||||
|
||||
@@ -49,7 +49,6 @@ func newRefererRoutes(t *testing.T) (http.Handler, *Handlers, *countingFetcher)
|
||||
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
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())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -158,7 +158,6 @@ func DefaultConfig() *Config {
|
||||
contentTypeWebP,
|
||||
contentTypeAVIF,
|
||||
contentTypeJXL,
|
||||
contentTypeSVG,
|
||||
},
|
||||
AllowHTTP: false,
|
||||
MaxConnectionsPerHost: DefaultMaxConnectionsPerHost,
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -133,10 +133,26 @@ var ErrInputTooLarge = errors.New("input image dimensions exceed maximum")
|
||||
// configured byte limit.
|
||||
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
|
||||
// not supported.
|
||||
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
|
||||
// processed and none finishes within ProcessingWaitTimeout.
|
||||
var ErrTooManyImages = errors.New("too many images being processed at once")
|
||||
@@ -219,6 +235,17 @@ func (p *ImageProcessor) Process(
|
||||
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
|
||||
img, err := vips.NewImageFromBuffer(data)
|
||||
if err != nil {
|
||||
@@ -237,16 +264,16 @@ func (p *ImageProcessor) Process(
|
||||
origWidth := img.Width()
|
||||
origHeight := img.Height()
|
||||
|
||||
// Detect input format
|
||||
inputFormat := p.detectFormat(img)
|
||||
|
||||
// Validate input dimensions to prevent DoS via decompression bombs
|
||||
if origWidth > MaxInputDimension || origHeight > MaxInputDimension {
|
||||
return nil, ErrInputTooLarge
|
||||
}
|
||||
|
||||
// Determine target dimensions
|
||||
targetWidth, targetHeight := targetDimensions(req.Size, origWidth, origHeight)
|
||||
// Determine target dimensions; too large ones are refused before resizing
|
||||
targetWidth, targetHeight, err := targetDimensions(req.Size, origWidth, origHeight)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Resize if needed
|
||||
if targetWidth != origWidth || targetHeight != origHeight {
|
||||
@@ -259,7 +286,7 @@ func (p *ImageProcessor) Process(
|
||||
// orig is the source's own format; encode refuses an empty format
|
||||
outputFormat := req.Format
|
||||
if outputFormat == FormatOriginal {
|
||||
outputFormat = p.formatFromString(inputFormat)
|
||||
outputFormat = sourceFormat
|
||||
}
|
||||
|
||||
// Encode to target format
|
||||
@@ -282,21 +309,29 @@ func (p *ImageProcessor) Process(
|
||||
|
||||
// targetDimensions calculates the output dimensions for a requested size,
|
||||
// scaling proportionally when only one dimension is given and keeping the
|
||||
// original dimensions when both are zero.
|
||||
func targetDimensions(size Size, origWidth, origHeight int) (int, int) {
|
||||
// original dimensions when both are zero. A side scaled proportionally can
|
||||
// come out above MaxOutputDimension, which is ErrOutputTooLarge.
|
||||
func targetDimensions(size Size, origWidth, origHeight int) (int, int, error) {
|
||||
width, height := size.Width, size.Height
|
||||
|
||||
switch {
|
||||
case size.Width == 0 && size.Height == 0:
|
||||
case width == 0 && height == 0:
|
||||
// Both are 0: keep original size
|
||||
return origWidth, origHeight
|
||||
case size.Width == 0:
|
||||
return origWidth, origHeight, nil
|
||||
case width == 0:
|
||||
// Only height specified: calculate width proportionally
|
||||
return origWidth * size.Height / origHeight, size.Height
|
||||
case size.Height == 0:
|
||||
width = origWidth * height / origHeight
|
||||
case height == 0:
|
||||
// Only width specified: calculate height proportionally
|
||||
return size.Width, origHeight * size.Width / origWidth
|
||||
default:
|
||||
return size.Width, size.Height
|
||||
height = origHeight * width / origWidth
|
||||
}
|
||||
|
||||
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.
|
||||
@@ -403,10 +438,9 @@ func (p *ImageProcessor) acquireSlot(ctx context.Context) (func(), error) {
|
||||
}
|
||||
}
|
||||
|
||||
// detectFormat returns the format string from a vips image.
|
||||
func (p *ImageProcessor) detectFormat(img *vips.ImageRef) string {
|
||||
format := img.Format()
|
||||
switch format {
|
||||
// detectFormat returns the format string for a govips image type.
|
||||
func (p *ImageProcessor) detectFormat(imageType vips.ImageType) string {
|
||||
switch imageType {
|
||||
case vips.ImageTypeJPEG:
|
||||
return "jpeg"
|
||||
case vips.ImageTypePNG:
|
||||
@@ -428,7 +462,9 @@ func (p *ImageProcessor) detectFormat(img *vips.ImageRef) string {
|
||||
}
|
||||
}
|
||||
|
||||
// resize resizes the image according to the fit mode.
|
||||
// resize resizes the image according to the fit mode. With FitOutside one
|
||||
// side can come out larger than requested; when it would be above
|
||||
// MaxOutputDimension, resize returns ErrOutputTooLarge without resizing.
|
||||
func (p *ImageProcessor) resize(
|
||||
img *vips.ImageRef, width, height int, fit FitMode,
|
||||
) error {
|
||||
@@ -476,6 +512,12 @@ func (p *ImageProcessor) resize(
|
||||
newW := int(float64(imgW) * 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)
|
||||
|
||||
default:
|
||||
@@ -687,22 +729,23 @@ func jxlDistance(quality int) float64 {
|
||||
return 53.0/3000.0*q*q - 23.0/20.0*q + 25.0
|
||||
}
|
||||
|
||||
// formatFromString converts a format string to Format.
|
||||
func (p *ImageProcessor) formatFromString(format string) Format {
|
||||
// formatFromString converts a format string from detectFormat to Format. A
|
||||
// format detectFormat could not name is ErrUnsupportedInputFormat.
|
||||
func (p *ImageProcessor) formatFromString(format string) (Format, error) {
|
||||
switch format {
|
||||
case "jpeg":
|
||||
return FormatJPEG
|
||||
return FormatJPEG, nil
|
||||
case "png":
|
||||
return FormatPNG
|
||||
return FormatPNG, nil
|
||||
case "gif":
|
||||
return FormatGIF
|
||||
return FormatGIF, nil
|
||||
case "webp":
|
||||
return FormatWebP
|
||||
return FormatWebP, nil
|
||||
case string(FormatAVIF):
|
||||
return FormatAVIF
|
||||
return FormatAVIF, nil
|
||||
case string(FormatJXL):
|
||||
return FormatJXL
|
||||
return FormatJXL, nil
|
||||
default:
|
||||
return FormatJPEG
|
||||
return "", fmt.Errorf("%w: %s", ErrUnsupportedInputFormat, format)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
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))
|
||||
}
|
||||
}
|
||||
}
|
||||
+83
-15
@@ -11,6 +11,8 @@ import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
lru "github.com/hashicorp/golang-lru/v2"
|
||||
@@ -33,7 +35,6 @@ const metaCacheSize = 10000
|
||||
// CacheConfig holds cache configuration.
|
||||
type CacheConfig struct {
|
||||
StateDir string
|
||||
CacheTTL time.Duration
|
||||
NegativeTTL time.Duration
|
||||
|
||||
// MaxBytes is the disk cache size limit in bytes that eviction
|
||||
@@ -79,6 +80,27 @@ type Cache struct {
|
||||
evictionDone chan struct{}
|
||||
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
|
||||
// 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.
|
||||
@@ -139,6 +161,9 @@ func newCache(
|
||||
metaCache: metaCache,
|
||||
contentLocks: newContentLock(),
|
||||
|
||||
pendingCountsAdded: make(chan struct{}, 1),
|
||||
pendingCountsDone: make(chan struct{}),
|
||||
|
||||
reconciliationPageSize: defaultReconciliationPageSize,
|
||||
}
|
||||
|
||||
@@ -474,12 +499,7 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
|
||||
func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
|
||||
var stats CacheStats
|
||||
|
||||
// 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)
|
||||
|
||||
err := c.readHitAndMissCounts(ctx, &stats)
|
||||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, fmt.Errorf("failed to get cache stats: %w", err)
|
||||
}
|
||||
@@ -511,19 +531,30 @@ func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
|
||||
}
|
||||
|
||||
// IncrementStats counts a cache hit or miss, and an upstream fetch that read
|
||||
// fetchBytes bytes, as IncrementUpstreamFetch does.
|
||||
// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment
|
||||
// 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) {
|
||||
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
|
||||
defer cancel()
|
||||
|
||||
var err error
|
||||
|
||||
pendingCount := &c.pendingMisses
|
||||
|
||||
if hit {
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
pendingCount = &c.pendingHits
|
||||
|
||||
_, err = c.db.ExecContext(countCtx, `
|
||||
UPDATE cache_stats
|
||||
SET hit_count = hit_count + 1,
|
||||
last_updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = 1
|
||||
`)
|
||||
} else {
|
||||
_, err = c.db.ExecContext(ctx, `
|
||||
_, err = c.db.ExecContext(countCtx, `
|
||||
UPDATE cache_stats
|
||||
SET miss_count = miss_count + 1,
|
||||
last_updated_at = CURRENT_TIMESTAMP
|
||||
@@ -531,7 +562,10 @@ func (c *Cache) IncrementStats(ctx context.Context, hit bool, fetchBytes int64)
|
||||
`)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
switch {
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -545,14 +579,22 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
|
||||
return
|
||||
}
|
||||
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
|
||||
defer cancel()
|
||||
|
||||
_, err := c.db.ExecContext(countCtx, `
|
||||
UPDATE cache_stats
|
||||
SET upstream_fetch_count = upstream_fetch_count + 1,
|
||||
upstream_fetch_bytes = upstream_fetch_bytes + ?,
|
||||
last_updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = 1
|
||||
`, 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",
|
||||
"fetch_bytes", fetchBytes, "error", err)
|
||||
}
|
||||
@@ -560,17 +602,43 @@ func (c *Cache) IncrementUpstreamFetch(ctx context.Context, fetchBytes int64) {
|
||||
|
||||
// IncrementTransformCount counts one image transcoded by the image processor.
|
||||
func (c *Cache) IncrementTransformCount(ctx context.Context) {
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
countCtx, cancel := withoutCancelKeepingDeadline(ctx)
|
||||
defer cancel()
|
||||
|
||||
_, err := c.db.ExecContext(countCtx, `
|
||||
UPDATE cache_stats
|
||||
SET transform_count = transform_count + 1,
|
||||
last_updated_at = CURRENT_TIMESTAMP
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
// hold: it reads the content type from the variant's .meta file and
|
||||
// keeps it in metaCache, unless a StoreVariant has put one there
|
||||
|
||||
@@ -102,7 +102,6 @@ func setupTestCache(t *testing.T) *Cache {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
@@ -241,7 +240,6 @@ func TestCache_NegativeCacheExpiry(t *testing.T) {
|
||||
// Very short negative TTL for testing
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 1 * time.Millisecond,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -1,116 +0,0 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// holdDatabase takes the one connection of the test service's database, so
|
||||
// that every other query waits for it, and returns the func that frees it.
|
||||
func holdDatabase(t *testing.T, svc *Service) func() {
|
||||
t.Helper()
|
||||
|
||||
conn, err := svc.cache.db.Conn(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to take the database connection: %v", err)
|
||||
}
|
||||
|
||||
release := func() { _ = conn.Close() }
|
||||
t.Cleanup(release)
|
||||
|
||||
return release
|
||||
}
|
||||
|
||||
// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
|
||||
// database's one connection while a request whose fetch is held reaches its
|
||||
// deadline. The request must still return by its deadline with the
|
||||
// deadline's error, and its miss must be counted once the connection is free.
|
||||
func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
const timeout = 200 * time.Millisecond
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), timeout)
|
||||
defer cancel()
|
||||
|
||||
results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover))
|
||||
|
||||
// The request has made its database reads by the time it fetches.
|
||||
<-fetcher.started
|
||||
|
||||
releaseDatabase := holdDatabase(t, svc)
|
||||
|
||||
select {
|
||||
case got := <-results:
|
||||
deadline, _ := ctx.Deadline()
|
||||
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(timeout + time.Second):
|
||||
t.Fatal("request did not return by its deadline while the database was busy")
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
waitCtx, cancelWait := context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancelWait()
|
||||
|
||||
if !svc.WaitForCountWrites(waitCtx) {
|
||||
t.Fatal("the miss was not counted once the database was free")
|
||||
}
|
||||
|
||||
want := cacheStatsCounters{missCount: 1}
|
||||
if got := readCacheStatsCounters(t, svc.cache); got != want {
|
||||
t.Errorf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_WaitForCountWrites holds the database's one connection while a
|
||||
// request past its deadline counts a miss. WaitForCountWrites must report the
|
||||
// count unwritten when its context ends, and written once the connection is
|
||||
// free.
|
||||
func TestService_WaitForCountWrites(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, _, _ := setupHeldFetchService(t)
|
||||
|
||||
releaseDatabase := holdDatabase(t, svc)
|
||||
|
||||
ended, cancel := context.WithDeadline(t.Context(), time.Now())
|
||||
defer cancel()
|
||||
|
||||
svc.writeCount(ended, func(ctx context.Context) {
|
||||
svc.cache.IncrementStats(ctx, false, 0)
|
||||
})
|
||||
|
||||
shortCtx, cancelShort := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
defer cancelShort()
|
||||
|
||||
written := svc.WaitForCountWrites(shortCtx)
|
||||
t.Logf("WaitForCountWrites() while the database was busy = %t", written)
|
||||
|
||||
if written {
|
||||
t.Fatal("WaitForCountWrites() = true while the database was busy")
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
waitCtx, cancelWait := context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancelWait()
|
||||
|
||||
if !svc.WaitForCountWrites(waitCtx) {
|
||||
t.Fatal("WaitForCountWrites() = false once the database was free")
|
||||
}
|
||||
|
||||
want := cacheStatsCounters{missCount: 1}
|
||||
if got := readCacheStatsCounters(t, svc.cache); got != want {
|
||||
t.Errorf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
@@ -70,7 +70,6 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
|
||||
// backwards compatibility with existing fixtures.
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
MaxBytes: maxBytes,
|
||||
DisableDiskCache: maxBytes == 0,
|
||||
|
||||
@@ -4,6 +4,5 @@ import "time"
|
||||
|
||||
// CacheConfig defaults.
|
||||
const (
|
||||
DefaultCacheTTL = 24 * time.Hour
|
||||
DefaultNegativeTTL = 5 * time.Minute
|
||||
)
|
||||
|
||||
@@ -15,7 +15,6 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
@@ -63,7 +62,6 @@ func TestNegativeCache_Expired(t *testing.T) {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 1 * time.Millisecond, // very short TTL
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,327 @@
|
||||
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})
|
||||
}
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"runtime/debug"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/dustin/go-humanize"
|
||||
@@ -37,9 +36,6 @@ type Service struct {
|
||||
// variantsInProgress lets the requests that miss the same variant at the
|
||||
// same time share one fetch and one transcode.
|
||||
variantsInProgress singleflight.Group
|
||||
// countWrites holds the count writes writeCount has started, for
|
||||
// WaitForCountWrites.
|
||||
countWrites sync.WaitGroup
|
||||
}
|
||||
|
||||
// ServiceConfig holds configuration for the image service.
|
||||
@@ -165,9 +161,8 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
|
||||
s.log.Error("failed to get cached variant", "key", result.CacheKey, "error", err)
|
||||
// Fall through to re-process
|
||||
} else {
|
||||
s.writeCount(ctx, func(ctx context.Context) {
|
||||
s.cache.IncrementStats(ctx, true, 0)
|
||||
})
|
||||
// Counted also when the request context has ended meanwhile
|
||||
s.cache.IncrementStats(ctx, true, 0)
|
||||
|
||||
return &ImageResponse{
|
||||
Content: reader,
|
||||
@@ -184,9 +179,7 @@ func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, e
|
||||
// failed or the request context has ended meanwhile
|
||||
response, err := s.processOrWait(ctx, req)
|
||||
|
||||
s.writeCount(ctx, func(ctx context.Context) {
|
||||
s.cache.IncrementStats(ctx, false, 0)
|
||||
})
|
||||
s.cache.IncrementStats(ctx, false, 0)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -215,25 +208,6 @@ func (s *Service) WaitForProcessing(ctx context.Context) int {
|
||||
return s.processor.WaitForProcessing(ctx)
|
||||
}
|
||||
|
||||
// WaitForCountWrites waits until every count a request has started writing
|
||||
// to the database is written, or until ctx ends, and reports whether they
|
||||
// all were.
|
||||
func (s *Service) WaitForCountWrites(ctx context.Context) bool {
|
||||
written := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
s.countWrites.Wait()
|
||||
close(written)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-written:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// ValidateRequest validates the request signature if required.
|
||||
func (s *Service) ValidateRequest(req *ImageRequest) error {
|
||||
// Check if host is allowed (no signature required)
|
||||
@@ -280,38 +254,6 @@ func (s *Service) GenerateSignedURL(
|
||||
baseURL, path, sig, exp, req.Quality, req.FitMode), nil
|
||||
}
|
||||
|
||||
// writeCount runs write, which adds to a count in the database, in a
|
||||
// goroutine of its own. The write gets ctx without its cancellation or
|
||||
// deadline, so an ended request is still counted. writeCount waits for it
|
||||
// until ctx's deadline, even if ctx is cancelled first, so a request
|
||||
// returns with its count written unless its deadline has passed; past
|
||||
// that, it does not wait for the database connection, which every request
|
||||
// shares.
|
||||
func (s *Service) writeCount(ctx context.Context, write func(context.Context)) {
|
||||
written := make(chan struct{})
|
||||
|
||||
s.countWrites.Go(func() {
|
||||
defer close(written)
|
||||
|
||||
write(context.WithoutCancel(ctx))
|
||||
})
|
||||
|
||||
deadline, hasDeadline := ctx.Deadline()
|
||||
if !hasDeadline {
|
||||
<-written
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
deadlineTimer := time.NewTimer(time.Until(deadline))
|
||||
defer deadlineTimer.Stop()
|
||||
|
||||
select {
|
||||
case <-written:
|
||||
case <-deadlineTimer.C:
|
||||
}
|
||||
}
|
||||
|
||||
// errPanicked is returned when processing a variant panicked.
|
||||
var errPanicked = errors.New("panic while processing image")
|
||||
|
||||
@@ -358,14 +300,8 @@ func (s *Service) processOrWait(
|
||||
}
|
||||
}()
|
||||
|
||||
processingCtx := context.WithoutCancel(ctx)
|
||||
|
||||
if deadline, ok := ctx.Deadline(); ok {
|
||||
var cancel context.CancelFunc
|
||||
|
||||
processingCtx, cancel = context.WithDeadline(processingCtx, deadline)
|
||||
defer cancel()
|
||||
}
|
||||
processingCtx, cancel := withoutCancelKeepingDeadline(ctx)
|
||||
defer cancel()
|
||||
|
||||
return s.processFromSourceOrFetch(processingCtx, req, cacheKey)
|
||||
})
|
||||
@@ -402,6 +338,20 @@ func (s *Service) processOrWait(
|
||||
}, 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
|
||||
// returns it with its size; nil if the cached data is unavailable, empty or
|
||||
// exceeds maxResponseSize.
|
||||
@@ -510,9 +460,8 @@ func (s *Service) fetchAndProcess(
|
||||
sourceData, err := io.ReadAll(fetchResult.Content)
|
||||
fetchBytes := int64(len(sourceData))
|
||||
|
||||
s.writeCount(ctx, func(ctx context.Context) {
|
||||
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
|
||||
})
|
||||
// Counted also when the request context has ended meanwhile
|
||||
s.cache.IncrementUpstreamFetch(ctx, fetchBytes)
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read upstream response: %w", err)
|
||||
@@ -593,7 +542,8 @@ func (s *Service) processAndStore(
|
||||
|
||||
processDuration := time.Since(processStart)
|
||||
|
||||
s.writeCount(ctx, s.cache.IncrementTransformCount)
|
||||
// Counted also when the request context has ended meanwhile
|
||||
s.cache.IncrementTransformCount(ctx)
|
||||
|
||||
// Read processed content
|
||||
processedData, err := io.ReadAll(processResult.Content)
|
||||
|
||||
@@ -2,7 +2,9 @@ package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -430,6 +432,172 @@ 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) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -43,7 +43,6 @@ func TestStats_HitRateIsRatio(t *testing.T) {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
@@ -91,7 +90,6 @@ func TestStats_ZeroCounts(t *testing.T) {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
@@ -120,7 +118,6 @@ func TestStats_LogsFailedCountQueries(t *testing.T) {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
|
||||
})
|
||||
@@ -162,7 +159,6 @@ func TestIncrementStats_LogsFailedUpdates(t *testing.T) {
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
|
||||
})
|
||||
@@ -232,7 +228,6 @@ func TestStats_DisabledCacheReportsNoItems(t *testing.T) {
|
||||
|
||||
disabled, err := NewCache(enabled.db, CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
DisableDiskCache: true,
|
||||
})
|
||||
|
||||
@@ -185,7 +185,6 @@ func SetupTestService(
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
CacheTTL: time.Hour,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
)
|
||||
|
||||
// URL parsing errors.
|
||||
@@ -20,8 +22,9 @@ var (
|
||||
ErrPathTraversal = errors.New("path traversal detected")
|
||||
)
|
||||
|
||||
// MaxDimension is the maximum allowed width or height.
|
||||
const MaxDimension = 8192
|
||||
// MaxDimension is the maximum allowed width or height, the same as the image
|
||||
// processor's maximum for an output image.
|
||||
const MaxDimension = imageprocessor.MaxOutputDimension
|
||||
|
||||
// ValidateDimension checks one requested width or height; name ("width" or
|
||||
// "height") appears in the error. 0 means "original size" and is valid.
|
||||
|
||||
@@ -98,9 +98,10 @@ func (s *Middleware) ClientIP() func(http.Handler) http.Handler {
|
||||
// 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.
|
||||
// Counts are kept only for the current and the previous window, so memory
|
||||
// stays bounded.
|
||||
// stays bounded. limitHandler writes the 429 after Retry-After is set; nil
|
||||
// keeps the rate limit library's plain-text "Too Many Requests".
|
||||
func (s *Middleware) RateLimit(
|
||||
requestLimit int, window time.Duration,
|
||||
requestLimit int, window time.Duration, limitHandler http.HandlerFunc,
|
||||
) func(http.Handler) http.Handler {
|
||||
return httprate.LimitBy(requestLimit, window,
|
||||
func(r *http.Request) (string, error) {
|
||||
@@ -112,7 +113,8 @@ func (s *Middleware) RateLimit(
|
||||
}
|
||||
|
||||
return httprate.CanonicalizeIP(ip), nil
|
||||
})
|
||||
},
|
||||
httprate.WithLimitHandler(limitHandler))
|
||||
}
|
||||
|
||||
// requestIDPattern is what a request's own X-Request-Id must look like to be
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,204 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -66,7 +66,7 @@ func (s *Server) SetupRoutes() {
|
||||
r.Use(s.h.LimitBody(handlers.MaxFormBytes))
|
||||
r.Use(s.h.CSRF())
|
||||
r.Get("/", s.h.HandleRoot())
|
||||
r.With(s.mw.RateLimit(LoginAttemptsPerMinute, time.Minute)).
|
||||
r.With(s.mw.RateLimit(LoginAttemptsPerMinute, time.Minute, nil)).
|
||||
Post("/", s.h.HandleRoot())
|
||||
r.Post("/generate", s.h.HandleGenerateURL())
|
||||
})
|
||||
@@ -88,6 +88,15 @@ func (s *Server) SetupRoutes() {
|
||||
r.Group(func(r chi.Router) {
|
||||
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
|
||||
// /v1/image/<host>/<path>/<width>x<height>.<format>, or with no
|
||||
// format /v1/image/<host>/<path>/<width>x<height>
|
||||
|
||||
@@ -29,10 +29,6 @@ const (
|
||||
// still being processed once ShutdownTimeout has passed.
|
||||
var errStillProcessing = errors.New("images still being processed at shutdown")
|
||||
|
||||
// errStillWritingCounts is returned by the server's stop hook when counts
|
||||
// are still being written to the database once ShutdownTimeout has passed.
|
||||
var errStillWritingCounts = errors.New("counts still being written at shutdown")
|
||||
|
||||
// Params defines dependencies for Server.
|
||||
type Params struct {
|
||||
fx.In
|
||||
@@ -121,11 +117,9 @@ func (s *Server) enableSentry() error {
|
||||
}
|
||||
|
||||
// cleanShutdown stops the HTTP server, waits for the images still being
|
||||
// processed and then for the counts still being written to the database,
|
||||
// which closes after this hook, then flushes Sentry. The first three share
|
||||
// ShutdownTimeout. It returns errStillProcessing and errStillWritingCounts
|
||||
// for the images and counts still unfinished after that, as their work is
|
||||
// abandoned.
|
||||
// processed, then flushes Sentry. The first two share ShutdownTimeout. It
|
||||
// returns errStillProcessing when images are still being processed after
|
||||
// that, as their work is abandoned.
|
||||
func (s *Server) cleanShutdown(ctx context.Context) error {
|
||||
s.log.Info("shutting down")
|
||||
|
||||
@@ -138,26 +132,17 @@ func (s *Server) cleanShutdown(ctx context.Context) error {
|
||||
}
|
||||
|
||||
stillProcessing := s.h.WaitForProcessing(ctxShutdown)
|
||||
countsWritten := s.h.WaitForCountWrites(ctxShutdown)
|
||||
|
||||
if s.sentryEnabled {
|
||||
sentry.Flush(SentryFlushTimeout)
|
||||
}
|
||||
|
||||
var unfinished []error
|
||||
|
||||
if stillProcessing > 0 {
|
||||
s.log.Error("images still being processed at shutdown",
|
||||
"count", stillProcessing)
|
||||
|
||||
unfinished = append(unfinished, errStillProcessing)
|
||||
return errStillProcessing
|
||||
}
|
||||
|
||||
if !countsWritten {
|
||||
s.log.Error("counts still being written at shutdown")
|
||||
|
||||
unfinished = append(unfinished, errStillWritingCounts)
|
||||
}
|
||||
|
||||
return errors.Join(unfinished...)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
+3
-1
@@ -94,7 +94,8 @@ main() {
|
||||
}
|
||||
|
||||
# start_containers starts a new origin and a new pixad, and waits up to 30
|
||||
# seconds for pixad's health check to pass.
|
||||
# seconds for pixad's health check to pass. pixad sees all of vegeta's
|
||||
# requests come from one address, so its image request limit is off.
|
||||
start_containers() {
|
||||
docker run -d --name "$run-origin" \
|
||||
--network "$run" --network-alias origin \
|
||||
@@ -104,6 +105,7 @@ start_containers() {
|
||||
--network "$run" -p 127.0.0.1::8080 --health-interval=1s \
|
||||
-e PIXA_SIGNING_KEY="$(head -c 32 /dev/urandom | base64)" \
|
||||
-e PIXA_ALLOWLIST_HOSTS=origin -e PIXA_ALLOW_HTTP=true \
|
||||
-e PIXA_IMAGE_REQUESTS_PER_MINUTE=0 \
|
||||
"$image" >/dev/null
|
||||
|
||||
waited=0
|
||||
|
||||
Reference in New Issue
Block a user