Author SHA1 Message Date
clawbot 1ae38a2425 Say govips gives pixa no way to stop an encode; give AVIF's worst case
check / check (push) Waiting to run
libvips can stop a running encode, but govips gives pixa no way to ask
it to, which README.md "Routes" and the Process and encode comments
now say in place of "libvips cannot stop an encode". The avifEffort
comment gives its 51 seconds as the RGB time and 110 seconds with an
alpha channel as the worst case, as README.md does.

Model: opus-5-5
2026-10-15 12:20:50 +00:00
clawbot 5101b9c4de Answer 503 to a request waiting on a transcode stopped at a deadline
A request that waited for another request's transcode got that
request's deadline error when the transcode stopped at its deadline,
and both image routes answered it 500, though the waiting request's
own downstream_timeout had time left. The image service now returns
ErrProcessingTimedOut for it, which both routes answer 503 "server
busy, try again later", as when pixa is busy. README.md "Routes" says
so.

Model: opus-5-5
2026-10-15 12:20:50 +00:00
clawbot a453aa9798 Test the 503 for a request waiting on a transcode stopped at a deadline
A request that waited for another request's transcode, stopped at that
request's deadline, must get ErrProcessingTimedOut, which both image
routes answer 503. Declares that error so the tests build. Also tests
that Process, past its deadline before it starts, never reads its input.

Model: opus-5-5
2026-10-15 12:20:50 +00:00
clawbot 1ba4c8d39f Stop a transcode whose request has ended before it encodes (closes #243)
Process looked at its context only while it waited for a processing
slot, so a transcode went on decoding and encoding, holding its slot,
however long after its request's downstream_timeout that took.
acquireSlot now gives the slot back and returns the context's error
when the context has ended by the time it has a slot, and encode
returns that error without beginning. libvips cannot stop an encode
once it has begun, so that encode keeps its slot until it finishes.
README.md gives how long that can take per output format at the 8192
limits and no longer says the work stops when downstream_timeout ends;
TODO.md is corrected the same way.

Model: opus-5-5
2026-10-15 12:00:34 +00:00
clawbot 74ce122dc4 Test that Process stops once its deadline has passed
Process looks at its context only while it waits for a processing
slot, so a request whose deadline has passed, before Process starts or
while it reads its input, still has its image decoded and encoded while
it holds the slot. The new test fails until Process returns
context.DeadlineExceeded in both cases and gives the slot back
(#243).

Model: opus-5-5
2026-10-15 12:00:34 +00:00
clawbot 2790125cb2 Serve a cached variant while its source URL is in the negative cache (closes #245)
check / check (push) Waiting to run
Get checked the negative cache before the variant cache, so after one
failed fetch of a source every variant of it answered 502 for five
minutes, also one on disk or one that could be made from the cached
source. The check now sits at the start of fetchAndProcess, which runs
only when no cached source can be used. A refusal from the negative
cache now counts as a miss.

Model: opus-5-5
2026-10-15 13:45:25 +02:00
clawbot c23585dd6d Lower-case the image URL's host and require a signature for a port (closes #246)
check / check (push) Waiting to run
With `cdn.example.com` allowlisted, other letter cases and a host with a port also passed the allowlist; each spelling was fetched and cached on its own with its own per-host connection limit, and a port let an unsigned URL reach any port of the host.

The image URL parser now lower-cases the host and refuses one holding `@` with 400. A host followed by `:`, with or without a port (read as Go's URL parser reads it, unbracketed IPv6 included), never counts as allowlisted, so it needs a signature. The host is signed lower-cased, as the URL writes it, so a signature made over an upper-case spelling is refused with 401.

Model: opus-5-5
2026-10-15 12:31:14 +02:00
17 changed files with 586 additions and 184 deletions
+38 -25
View File
@@ -126,17 +126,6 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
default 75% of (free space + what the cache holds)
- the rest are in the table under Configuration below
- **Memory:** pixa can use up to `max_concurrent_processing` times 1.6 GiB for
the images it processes at once, plus `upstream_connections` times
`upstream_max_response_size` for the source images it has fetched that wait
for processing. 1.6 GiB is the most pixad's resident memory was measured to
grow while it processed one 8192x8192 source, the largest it accepts, into
each output format in turn, one at a time; it grew most for a 40 MB JPEG of
random pixels saved as AVIF. At the defaults that is half the CPUs pixa sees
times 1.6 GiB, plus 64 times 50 MiB: on a host where pixa sees 48 CPUs, 24
times 1.6 GiB plus 3.1 GiB, about 42 GiB. Set `PIXA_MAX_CONCURRENT_PROCESSING`
and `PIXA_UPSTREAM_CONNECTIONS` so that this fits the memory the container may
use.
- **Health check:** the image's `HEALTHCHECK` requests
`/.well-known/healthcheck.json`. upaas reads the container's health 60 seconds
after a deploy and marks the deploy failed unless it is `healthy`. The probe
@@ -211,9 +200,10 @@ path under `/v1/` answers 200, in maintenance mode too.
`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.
answered with an error status, and for 5 minutes after that for an image of
the same source URL that pixa can neither serve from its cache nor make from
the source image still in its cache; 503 when pixa is busy or in maintenance
mode; 500 for any other failure.
- `GET` or `HEAD` `/v1/e/<token>/<name>` — an image through an encrypted URL
(see Encrypted URLs). Needs: nothing but the URL. Answers: 200; 304 when
`If-None-Match` matches the image's `ETag`; 400 for a token that does not
@@ -324,9 +314,26 @@ cache at once, they share one upstream fetch (or one read of the cached source)
and one transcode: the first request does the work, and the others wait for its
image or its error, holding no upstream connection or processing slot of their
own. A waiting request stops waiting when its own client goes away. The work
goes on for the others even if the first request's client goes away, until that
request's `downstream_timeout` ends. The shared fetch sends the first request's
ID upstream, and the lines logged for the fetch and the transcode carry that ID.
goes on for the others even if the first request's client goes away. The shared
fetch sends the first request's ID upstream, and the lines logged for the fetch
and the transcode carry that ID.
A transcode whose request's `downstream_timeout` has ended (for a shared one,
the first request's) stops waiting for a processing slot, or, once it has one,
gives it back without beginning to encode. A request waiting for a shared
transcode that stops this way is answered 503 with the error
`server busy, try again later`, though its own `downstream_timeout` may not have
ended. govips gives pixa no way to stop libvips once an encode has begun, so an
encode keeps its slot until it finishes, which can be after
`downstream_timeout`. The limit of 8192 on each side of the source and of the
output bounds how long that takes. With one libvips thread, as pixad runs it,
encoding 8192x8192 random pixels takes up to about 1 second as JPEG, 15 as PNG,
16 as GIF, 63 as WebP, 110 as AVIF and 96 as JPEG XL, the slowest of 8 or 16
bits per sample, with or without an alpha channel. WebP, AVIF and JPEG XL are
that slow only with an alpha channel: 8-bit RGB takes about 44, 53 and 28
seconds. Decoding such a source adds up to about 11 seconds. A WebP, AVIF or
JPEG XL encode can therefore hold its slot past the default `downstream_timeout`
of 60 seconds.
The login form (`POST /`) is limited to 5 attempts per minute per client
address, counting an IPv6 client by its /64; an attempt over the limit is
@@ -429,7 +436,9 @@ HMAC-SHA256(secret, "host:path:query:width:height:format:expiration:quality:fit"
Where:
- `host` — source origin hostname (e.g. `cdn.example.com`)
- `host` — source host as the URL writes it up to the first `/`, in lower case,
keeping an IPv6 address's brackets and any `:` and port after it (e.g.
`cdn.example.com`, `cdn.example.com:8443`, `[2001:db8::1]:8443`)
- `path` — source path (e.g. `/photos/cat.jpg`), with each `%` written as `%25`
and each `:` as `%3A`, so `/photos/a:b.jpg` is signed as `/photos/a%3Ab.jpg`
- `query` — source query string, empty string if none, with `%` and `:` written
@@ -484,6 +493,12 @@ entry that is neither a host name (letters, digits, hyphens, underscores and
dots, with at most one leading dot) nor an IP address, such as one with a port
or a `*.` wildcard, aborts startup.
pixa lower-cases the host of a `/v1/image/` URL before anything else uses it, so
`CDN.Example.com` is matched, fetched, cached and counted against
`upstream_connections_per_host` as `cdn.example.com`. Such a URL whose host has
a port, as in `cdn.example.com:8443`, or a `:` with no port, needs a signature
even when the host is allowlisted. A host holding `@` is refused with 400.
### Configuration
Every setting can be given as an environment variable, in a YAML config file
@@ -526,7 +541,7 @@ tell. With no file, pixa uses the environment and the defaults.
| `PIXA_ALLOW_HTTP` | `allow_http` | Allow plain-HTTP upstreams, for testing only; default `false` |
| `PIXA_UPSTREAM_CONNECTIONS_PER_HOST` | `upstream_connections_per_host` | Concurrent connections per upstream host; default `20` |
| `PIXA_UPSTREAM_CONNECTIONS` | `upstream_connections` | Concurrent connections to all upstream hosts together; default `64` |
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default half the CPUs, at least 1 |
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default the number of CPUs |
| `PIXA_IMAGE_REQUESTS_PER_MINUTE` | `image_requests_per_minute` | Image requests per client per minute; `0` turns it off; default `600` |
| `PIXA_UPSTREAM_FETCH_TIMEOUT` | `upstream_fetch_timeout` | Time allowed for one fetch from an upstream host; default `30s` |
| `PIXA_UPSTREAM_MAX_RESPONSE_SIZE` | `upstream_max_response_size` | Largest upstream response accepted, in bytes; default 50 MiB |
@@ -612,12 +627,10 @@ Key settings in more detail:
does, and `downstream_timeout` has not ended first, the request is answered
503 with the error `server busy, try again later`
- `max_concurrent_processing` — the most images decoded and encoded at once;
default half the CPUs pixa can use (`GOMAXPROCS`, which follows a container's
CPU limit), rounded down, and at least 1. A value that is not a whole number
of at least 1 aborts startup. 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. Each one can take a lot of
memory: see "Running under upaas"
default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
container's CPU limit. A request that finds all of them in use waits up to 10
seconds for one to free up; if none does, 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.
+44 -15
View File
@@ -30,16 +30,45 @@ P2: security: per-origin rate limiting
# Completed Steps
- 2026-10-15 `max_concurrent_processing` defaults to half the CPUs pixa sees,
rounded down, and at least 1 (closes #242): it defaulted to one image per CPU,
and a container with no CPU limit sees every CPU of the host, so on a 48-CPU
host pixa processed 48 images at once, each of which can take about 1.6 GiB.
`imageprocessor.DefaultMaxConcurrentProcessing` gives the default to the
config and to `imageprocessor.New`; `PIXA_MAX_CONCURRENT_PROCESSING` still
overrides it. "Running under upaas" in `README.md` gives the worst-case
memory, measured, and tells the operator to set
`PIXA_MAX_CONCURRENT_PROCESSING` and `PIXA_UPSTREAM_CONNECTIONS` to fit the
container's memory.
- 2026-10-15 a transcode stops once its request's `downstream_timeout` has
ended, unless it has begun encoding (closes #243): the image processor looked
at its context only while it waited for a processing slot, so an image was
decoded and encoded, holding the slot, however long after its request's
deadline that took. `Process` now returns the context's error, and gives the
slot back, when the context has ended by the time it has the slot or before it
encodes. A request that waited for a shared transcode stopped at the first
request's deadline is answered 503 `server busy, try again later`, not 500,
though its own `downstream_timeout` may not have ended. govips gives pixa no
way to stop libvips once an encode has begun, so that encode keeps its slot
until it finishes; the limit of 8192 on each side of the source and of the
output bounds how long that takes. With one libvips thread, 8192x8192 random
pixels take up to about 1 second to encode as JPEG, 15 as PNG, 16 as GIF, 63
as WebP, 110 as AVIF and 96 as JPEG XL, the last three with an alpha channel,
plus up to about 11 seconds to decode the source; `README.md` gives these
times and no longer says the work stops when `downstream_timeout` ends. The
settings of the WebP, AVIF and JPEG XL encoders, which can hold a slot past
the default `downstream_timeout` of 60 seconds, are unchanged.
- 2026-10-15 a cached variant is served while its source URL is in the negative
cache (closes #245): `Get` checked the negative cache before the variant
cache, so once a fetch of the source failed, as one does when another variant
misses after the source was evicted and the upstream is down, every variant of
that source answered 502 for 5 minutes, also one on disk or one that could be
made from the cached source. The negative cache is now checked only just
before a fetch from the upstream, so a cached variant is served and a variant
not in the cache is made from the cached source when there is one. `README.md`
says so under `/v1/image/`. A refusal from the negative cache now counts as a
miss, as any other request that misses the variant cache does.
- 2026-10-15 the host of a `/v1/image/` URL is lower-cased, and one with a port
needs a signature (closes #246): with `cdn.example.com` allowlisted,
`CDN.Example.com` and `cdn.example.com:8443` passed the allowlist too, and
each spelling was fetched and cached on its own and had connections of its own
under `upstream_connections_per_host`. The image URL parser now lower-cases
the host and refuses one holding `@`, which a URL reads as a user name and
password, with 400. A host with a port, or with a `:` and no port, needs a
signature even when it is allowlisted. That is checked in the image service,
not in `allowlist.IsAllowed`, which also matches `referer_blocklist`, where a
`Referer` is blocked whatever its port. A URL is signed over its host as it
writes it, in lower case, with any IPv6 brackets, `:` and port kept.
- 2026-10-15 a `:` in the source path or query no longer moves where the path
ends and the query begins (closes #247): the signed data and the cache key
joined host, path and query with `:` and no escaping, so a URL signed for
@@ -511,11 +540,11 @@ P2: security: per-origin rate limiting
cancellation but keeps its deadline, and the others wait for its image or
error holding no upstream connection or processing slot, and stop waiting when
their own context ends; the request doing the processing waits for it even
then, up to its deadline; a request whose context has already ended starts
nothing; each request counts one miss, and the processing counts its fetch and
transcode once; a panic while processing is reported to Sentry when
`sentry_dsn` is set and becomes an error for every waiting request instead of
stopping pixad; documented in `README.md`.
then, until it ends, which can be after its deadline; a request whose context
has already ended starts nothing; each request counts one miss, and the
processing counts its fetch and transcode once; a panic while processing is
reported to Sentry when `sentry_dsn` is set and becomes an error for every
waiting request instead of stopping pixad; documented in `README.md`.
- 2026-09-29 only the image routes send CORS headers (closes #98): the CORS
middleware, with the `access_control_allow_origin` origin, moved from the
router root onto a `/v1` subrouter holding `/v1/image/` and `/v1/e/`, where it
+4 -5
View File
@@ -108,11 +108,10 @@ upstream_connections_per_host: 20
# downstream_timeout has ended first.
upstream_connections: 64
# Maximum number of images decoded and encoded at once (default: half the
# number of CPUs pixa can use, at least 1; the number of CPUs follows a
# container's CPU limit). A request that finds none free waits up to 10
# seconds for one, and if none frees up it is answered 503, unless
# downstream_timeout has ended first.
# Maximum number of images decoded and encoded at once (default: the
# number of CPUs pixa can use, which follows a container's CPU limit). A
# request that finds none free waits up to 10 seconds for one, and if none
# frees up it is answered 503, unless downstream_timeout has ended first.
# max_concurrent_processing: 4
# How many requests to the image routes (/v1/image/ and /v1/e/) one client
@@ -1,7 +1,6 @@
package config
import (
"fmt"
"runtime"
"testing"
)
@@ -13,8 +12,8 @@ const (
)
// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
// max_concurrent_processing is half the CPUs Go uses, and at least 1, and
// an omitted upstream_connections is 64.
// max_concurrent_processing is the number of CPUs Go uses and an omitted
// upstream_connections is 64.
func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
t.Parallel()
@@ -23,9 +22,9 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
t.Fatalf("minimal config should be valid, got error: %v", err)
}
if c.MaxConcurrentProcessing != max(runtime.GOMAXPROCS(0)/2, 1) {
t.Errorf("MaxConcurrentProcessing = %d, want %d, half the CPUs",
c.MaxConcurrentProcessing, max(runtime.GOMAXPROCS(0)/2, 1))
if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
}
if c.UpstreamConnections != 64 {
@@ -33,36 +32,6 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
}
}
// TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne checks
// that an omitted max_concurrent_processing is half the CPUs Go uses,
// rounded down, and 1 when Go uses one CPU.
//
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
func TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
cpusBefore := runtime.GOMAXPROCS(0)
t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
for _, tc := range []struct{ cpus, want int }{
{1, 1}, {2, 1}, {3, 1}, {8, 4},
} {
runtime.GOMAXPROCS(tc.cpus)
c, err := configFromYAML(t, signingKeyLine)
if err != nil {
t.Fatalf("minimal config should be valid, got error: %v", err)
}
t.Logf("GOMAXPROCS %d: MaxConcurrentProcessing %d",
tc.cpus, c.MaxConcurrentProcessing)
if c.MaxConcurrentProcessing != tc.want {
t.Errorf("GOMAXPROCS %d: MaxConcurrentProcessing = %d, want %d",
tc.cpus, c.MaxConcurrentProcessing, tc.want)
}
}
}
// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
// two limits are used as given.
func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
@@ -191,19 +160,3 @@ func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
})
}
}
// TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup checks
// that a PIXA_MAX_CONCURRENT_PROCESSING that is not a whole number of at
// least 1, an empty one included, aborts startup naming the variable and
// the key instead of taking the default.
func TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup(t *testing.T) {
for _, value := range []string{"", " ", "0.5", "half", "-1", "4 images"} {
t.Run(fmt.Sprintf("value %q", value), func(t *testing.T) {
t.Setenv(testMaxConcurrentProcessingVar, value)
_, err := configFromYAML(t, signingKeyLine)
wantStartupError(t, err,
testMaxConcurrentProcessingVar, keyMaxConcurrentProcessing)
})
}
}
+5 -3
View File
@@ -12,6 +12,7 @@ import (
"os"
"path/filepath"
"regexp"
"runtime"
"sort"
"strconv"
"strings"
@@ -20,7 +21,6 @@ import (
"git.eeqj.de/sneak/smartconfig"
"go.uber.org/fx"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/logger"
)
@@ -297,8 +297,10 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
UpstreamConnections: loader.intVal(
keyUpstreamConnections, DefaultUpstreamConnections),
MaxConcurrentProcessing: loader.intVal(keyMaxConcurrentProcessing,
imageprocessor.DefaultMaxConcurrentProcessing()),
// Decoding and encoding are CPU-bound, so the default is one image
// per CPU Go uses, which follows a container's CPU limit.
MaxConcurrentProcessing: loader.intVal(
keyMaxConcurrentProcessing, runtime.GOMAXPROCS(0)),
UpstreamFetchTimeout: loader.durationVal(
keyUpstreamFetchTimeout, DefaultUpstreamFetchTimeout),
UpstreamMaxResponseSize: loader.int64Val(
+2 -1
View File
@@ -240,7 +240,8 @@ func (s *Handlers) respondImageError(
}
if errors.Is(err, httpfetcher.ErrTooManyConnections) ||
errors.Is(err, imageprocessor.ErrTooManyImages) {
errors.Is(err, imageprocessor.ErrTooManyImages) ||
errors.Is(err, imgcache.ErrProcessingTimedOut) {
s.respondError(w, "server busy, try again later",
http.StatusServiceUnavailable)
+2 -1
View File
@@ -158,7 +158,8 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
case errors.Is(err, httpfetcher.ErrUpstreamTimeout):
s.respondError(w, "upstream timeout", http.StatusGatewayTimeout)
case errors.Is(err, httpfetcher.ErrTooManyConnections),
errors.Is(err, imageprocessor.ErrTooManyImages):
errors.Is(err, imageprocessor.ErrTooManyImages),
errors.Is(err, imgcache.ErrProcessingTimedOut):
s.respondError(w, "server busy, try again later",
http.StatusServiceUnavailable)
case errors.Is(err, httpfetcher.ErrInvalidContentType),
@@ -1,6 +1,7 @@
package handlers
import (
"context"
"fmt"
"log/slog"
"net/http"
@@ -47,3 +48,40 @@ func TestServerBusyAnswers503(t *testing.T) {
}
}
}
// TestProcessingTimedOutAnswers503 checks that respondImageError and
// handleImageError, which answer the errors of the two image routes, answer
// 503 with the server busy error when the image service reports that
// processing stopped at the deadline of the request doing it, wrapped as the
// service wraps it. A request that waited for that processing gets this error
// while its own deadline may have time left. The request's source only goes
// into a log line, so it is left empty.
func TestProcessingTimedOutAnswers503(t *testing.T) {
t.Parallel()
h := &Handlers{log: slog.New(slog.DiscardHandler)}
err := fmt.Errorf("%w: failed to encode: %w",
imgcache.ErrProcessingTimedOut, context.DeadlineExceeded)
plain := httptest.NewRecorder()
h.respondImageError(plain, &imgcache.ImageRequest{}, err)
encrypted := httptest.NewRecorder()
h.handleImageError(encrypted, err)
for name, rec := range map[string]*httptest.ResponseRecorder{
"respondImageError": plain, "handleImageError": encrypted,
} {
t.Logf("%s: status = %d, body = %q", name, rec.Code, rec.Body.String())
if rec.Code != http.StatusServiceUnavailable {
t.Errorf("%s: status = %d, want %d",
name, rec.Code, http.StatusServiceUnavailable)
}
if !strings.Contains(rec.Body.String(), "server busy, try again later") {
t.Errorf("%s: body = %q, want the server busy error",
name, rec.Body.String())
}
}
}
@@ -0,0 +1,106 @@
package imageprocessor
import (
"bytes"
"context"
"errors"
"io"
"testing"
"time"
)
// readerAfterDeadline is a Process input whose first Read waits until done
// is closed, so the deadline passes after Process has taken its slot and
// before it encodes.
type readerAfterDeadline struct {
data *bytes.Reader
done <-chan struct{}
waited bool
}
func (r *readerAfterDeadline) Read(p []byte) (int, error) {
if !r.waited {
r.waited = true
<-r.done
}
return r.data.Read(p)
}
// inputNotToBeRead is a Process input that fails the test when it is read.
type inputNotToBeRead struct {
t *testing.T
}
func (r inputNotToBeRead) Read([]byte) (int, error) {
r.t.Error("Process read its input although its deadline had passed")
return 0, io.EOF
}
// TestProcessStopsWhenDeadlinePassed checks that Process returns
// context.DeadlineExceeded, and holds no slot afterwards, when its context's
// deadline has passed before it starts, then without reading its input, or
// while it reads its input. govips gives pixa no way to stop libvips once an
// encode has begun, so Process must not begin one for a request that has
// ended.
func TestProcessStopsWhenDeadlinePassed(t *testing.T) {
t.Parallel()
input := createTestJPEG(t, 50, 50)
t.Run("before Process starts", func(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithDeadline(t.Context(),
time.Now().Add(-time.Second))
defer cancel()
// With a slot free, the select in acquireSlot takes the slot or sees
// the ended context at random, so Process is called many times
for range 100 {
checkProcessStopsAtDeadline(ctx, t, inputNotToBeRead{t: t})
}
})
t.Run("while Process reads its input", func(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
defer cancel()
checkProcessStopsAtDeadline(ctx, t, &readerAfterDeadline{
data: bytes.NewReader(input), done: ctx.Done(),
})
})
}
// checkProcessStopsAtDeadline runs Process on input with ctx, whose deadline
// has passed by the time Process could encode, and fails the test unless
// Process returns context.DeadlineExceeded and gives its slot back.
func checkProcessStopsAtDeadline(
ctx context.Context, t *testing.T, input io.Reader,
) {
t.Helper()
proc := New(Params{MaxConcurrentProcessing: 1})
result, err := proc.Process(ctx, input, smallJPEGRequest())
if result != nil {
t.Logf("Process() made a %dx%d image after its deadline passed",
result.Width, result.Height)
_ = result.Content.Close()
}
t.Logf("Process() error = %v", err)
if !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("Process() error = %v, want context.DeadlineExceeded", err)
}
if held := len(proc.processingSemaphore); held != 0 {
t.Errorf("%d slots still held after Process() returned", held)
}
}
+41 -36
View File
@@ -20,11 +20,11 @@ import (
var vipsOnce sync.Once
// initVips initializes libvips with quiet logging, one worker thread per
// image and no operation cache. With one thread each, the images processed
// at once use no more CPUs than MaxConcurrentProcessing. Each request
// decodes different source bytes, so the operation cache would rarely be
// hit and would hold memory outside MaxConcurrentProcessing; repeated
// requests are served from pixa's disk cache instead.
// image and no operation cache. Process already works on one image per CPU
// by default, so more threads per image would only compete for the CPUs.
// Each request decodes different source bytes, so the operation cache
// would rarely be hit and would hold memory outside MaxConcurrentProcessing;
// repeated requests are served from pixa's disk cache instead.
func initVips() {
vipsOnce.Do(func() {
vips.LoggingSettings(nil, vips.LogLevelError)
@@ -179,19 +179,10 @@ type Params struct {
// If <= 0, DefaultMaxInputBytes is used.
MaxInputBytes int64
// MaxConcurrentProcessing is the most images processed at once.
// If <= 0, DefaultMaxConcurrentProcessing() is used.
// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
MaxConcurrentProcessing int
}
// DefaultMaxConcurrentProcessing returns the most images processed at once
// when none is set: half the CPUs Go uses (runtime.GOMAXPROCS(0), which
// follows a container's CPU limit), rounded down, and at least 1.
func DefaultMaxConcurrentProcessing() int {
const cpusPerImage = 2
return max(runtime.GOMAXPROCS(0)/cpusPerImage, 1)
}
// New creates a new image processor with the given parameters.
// A zero-value Params{} uses sensible defaults.
func New(params Params) *ImageProcessor {
@@ -204,7 +195,7 @@ func New(params Params) *ImageProcessor {
maxConcurrentProcessing := params.MaxConcurrentProcessing
if maxConcurrentProcessing <= 0 {
maxConcurrentProcessing = DefaultMaxConcurrentProcessing()
maxConcurrentProcessing = runtime.GOMAXPROCS(0)
}
return &ImageProcessor{
@@ -217,7 +208,10 @@ func New(params Params) *ImageProcessor {
// Process transforms an image according to the request. When
// MaxConcurrentProcessing images are already being processed, it waits up
// to ProcessingWaitTimeout for one to finish, then fails with
// ErrTooManyImages.
// ErrTooManyImages. When ctx has ended once Process has a slot, or before it
// encodes, it returns ctx's error and gives the slot back. govips gives pixa
// no way to stop libvips once an encode has begun, so an encode keeps its
// slot until it finishes.
func (p *ImageProcessor) Process(
ctx context.Context,
input io.Reader,
@@ -299,7 +293,7 @@ func (p *ImageProcessor) Process(
}
// Encode to target format
output, err := p.encode(img, outputFormat, req.Quality)
output, err := p.encode(ctx, img, outputFormat, req.Quality)
if err != nil {
return nil, fmt.Errorf("failed to encode: %w", err)
}
@@ -426,25 +420,28 @@ func (p *ImageProcessor) WaitForProcessing(ctx context.Context) int {
// acquireSlot takes a slot in processingSemaphore, waiting at most
// processingWaitTimeout for one to free up, and returns the func that gives
// it back. A free slot is taken even when ctx has ended; only the wait for
// one stops when ctx ends, as the rest of Process does not check ctx.
// it back. When ctx has ended by the time it has a slot, it gives the slot
// back and returns ctx's error.
func (p *ImageProcessor) acquireSlot(ctx context.Context) (func(), error) {
release := func() { <-p.processingSemaphore }
select {
case p.processingSemaphore <- struct{}{}:
return release, nil
default:
}
select {
case p.processingSemaphore <- struct{}{}:
return release, nil
case <-time.After(p.processingWaitTimeout):
return nil, ErrTooManyImages
case <-ctx.Done():
return nil, ctx.Err()
}
release := func() { <-p.processingSemaphore }
// select takes the slot even when ctx has ended at the same moment
err := ctx.Err()
if err != nil {
release()
return nil, err
}
return release, nil
}
// detectFormat returns the format string for a govips image type.
@@ -536,10 +533,17 @@ func (p *ImageProcessor) resize(
const defaultQuality = 85
// encode encodes an image to the specified format.
// encode encodes an image to the specified format. govips gives pixa no way
// to stop libvips once an encode has begun, so when ctx has ended encode
// returns ctx's error without beginning one.
func (p *ImageProcessor) encode(
img *vips.ImageRef, format Format, quality int,
ctx context.Context, img *vips.ImageRef, format Format, quality int,
) ([]byte, error) {
err := ctx.Err()
if err != nil {
return nil, err
}
if quality <= 0 {
quality = defaultQuality
}
@@ -549,7 +553,7 @@ func (p *ImageProcessor) encode(
// libvips' built-in profile; govips' own sRGB path variable is set on
// first use but read without a lock, so concurrent requests race on it.
if img.HasICCProfile() {
err := img.TransformICCProfileWithFallback("srgb", "srgb")
err = img.TransformICCProfileWithFallback("srgb", "srgb")
if err != nil {
return nil, fmt.Errorf("failed to convert to sRGB: %w", err)
}
@@ -641,10 +645,11 @@ func exportWebP(img *vips.ImageRef, quality int) ([]byte, error) {
// avifEffort is the AVIF effort, from 0 (fastest) to 9; 1 is the lowest
// govips can set. With one thread, as pixad runs libvips, 1 takes about 51
// seconds to save an 8192x8192 image of random pixels, the worst case,
// against the default downstream_timeout of 60 seconds. On an image of
// milder noise, which 1 saves in about 12 seconds, 2 takes nearly a minute
// and libvips' default, 4, takes minutes.
// seconds to save an 8192x8192 RGB image of random pixels, and about 110
// with an alpha channel, the worst case, against the default
// downstream_timeout of 60 seconds. On an image of milder noise, which 1
// saves in about 12 seconds, 2 takes nearly a minute and libvips' default,
// 4, takes minutes.
const avifEffort = 1
// avifBitdepth is the AVIF bit depth, 8 bits per sample for every image.
@@ -115,14 +115,14 @@ func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
}
}
func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) {
func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
t.Parallel()
for _, limit := range []int{0, -1} {
proc := New(Params{MaxConcurrentProcessing: limit})
if got := cap(proc.processingSemaphore); got != max(runtime.GOMAXPROCS(0)/2, 1) {
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, half the CPUs",
limit, got, max(runtime.GOMAXPROCS(0)/2, 1))
if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
limit, got, runtime.GOMAXPROCS(0))
}
}
@@ -132,30 +132,6 @@ func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) {
}
}
// TestNewDefaultIsHalfTheCPUsAndAtLeastOne checks that New, given no
// MaxConcurrentProcessing, makes one slot for every two CPUs Go uses,
// rounded down, and one slot when Go uses one CPU.
//
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
func TestNewDefaultIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
cpusBefore := runtime.GOMAXPROCS(0)
t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
for _, tc := range []struct{ cpus, want int }{
{1, 1}, {2, 1}, {3, 1}, {8, 4},
} {
runtime.GOMAXPROCS(tc.cpus)
got := cap(New(Params{}).processingSemaphore)
t.Logf("GOMAXPROCS %d: %d slots", tc.cpus, got)
if got != tc.want {
t.Errorf("GOMAXPROCS %d: %d slots, want %d", tc.cpus, got, tc.want)
}
}
}
// TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
// than MaxConcurrentProcessing allows and holds the first ones inside
// Process until the test lets them go. No more than the limit may be
@@ -15,6 +15,7 @@ import (
"github.com/getsentry/sentry-go"
"sneak.berlin/go/pixa/internal/httpfetcher"
"sneak.berlin/go/pixa/internal/imageprocessor"
"sneak.berlin/go/pixa/internal/magic"
)
@@ -435,6 +436,47 @@ func TestService_Get_ReturnsByItsDeadline(t *testing.T) {
}
}
// TestService_Get_WaiterGetsProcessingTimedOut has a request with no deadline
// join the held fetch of a first request with a short one. With the only
// processing slot held, the first request's transcode waits for it until the
// first request's deadline ends, and stops then. The waiting request, whose
// own context has not ended, must get ErrProcessingTimedOut, which the image
// routes answer 503 as when pixa is busy.
func TestService_Get_WaiterGetsProcessingTimedOut(t *testing.T) {
t.Parallel()
svc, fixtures, fetcher := setupHeldFetchService(t)
svc.processor = imageprocessor.New(
imageprocessor.Params{MaxConcurrentProcessing: 1},
)
holdProcessingSlot(t, svc.processor)
const firstTimeout = 500 * time.Millisecond
firstCtx, cancelFirst := context.WithTimeout(t.Context(), firstTimeout)
defer cancelFirst()
first := startGet(firstCtx, svc, photoVariant(fixtures, 85, FitCover))
<-fetcher.started
waiter := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
time.Sleep(arrivalWait)
fetcher.releaseFetches()
t.Logf("first request: Get() error = %v", (<-first).err)
got := <-waiter
t.Logf("waiting request: Get() error = %v", got.err)
if !errors.Is(got.err, ErrProcessingTimedOut) {
t.Errorf("waiting request: Get() error = %v, want %v",
got.err, ErrProcessingTimedOut)
}
}
// panickingFetcher panics on every fetch.
type panickingFetcher struct{}
@@ -94,6 +94,71 @@ func TestNegativeCache_Expired(t *testing.T) {
}
}
// TestService_Get_ServesCachedVariantOfNegativeCachedURL caches a variant of
// the test photo, then keeps a failed fetch of the photo's URL in the negative
// cache, as a failed fetch for another variant does once the source has been
// evicted. The cached variant is still served, and a variant not in the cache
// is made from the cached source. Once the source is evicted, a variant not in
// the cache needs a fetch and is refused.
func TestService_Get_ServesCachedVariantOfNegativeCachedURL(t *testing.T) {
t.Parallel()
svc, fixtures := SetupTestService(t)
resp, err := svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
if err != nil {
t.Fatalf("Get() first request error = %v", err)
}
_ = resp.Content.Close()
err = svc.cache.StoreNegative(t.Context(), photoVariant(fixtures, 85, FitCover),
httpStatusBadGateway, "upstream answered 503")
if err != nil {
t.Fatalf("StoreNegative() error = %v", err)
}
resp, err = svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
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)
}
resp, err = svc.Get(t.Context(), photoVariant(fixtures, 40, FitCover))
t.Logf("Get() of a variant made from the cached source: error = %v", err)
if err != nil {
t.Fatalf("Get() of a variant made from the cached source error = %v", err)
}
_ = resp.Content.Close()
contentHash, _, err := svc.cache.LookupSource(t.Context(),
photoVariant(fixtures, 85, FitCover))
if err != nil || contentHash == "" {
t.Fatalf("LookupSource() = %q, %v, want the cached source", contentHash, err)
}
err = svc.cache.evictSourceBlob(t.Context(), contentHash)
if err != nil {
t.Fatalf("evictSourceBlob() error = %v", err)
}
_, err = svc.Get(t.Context(), photoVariant(fixtures, 60, FitCover))
t.Logf("Get() of a variant whose source was evicted: error = %v", err)
if !errors.Is(err, ErrNegativeCached) {
t.Errorf("Get() of a variant whose source was evicted error = %v, want %v",
err, ErrNegativeCached)
}
}
func TestService_Get_ReturnsErrorForNegativeCachedURL(t *testing.T) {
t.Parallel()
+44 -18
View File
@@ -9,6 +9,7 @@ import (
"log/slog"
"net/url"
"runtime/debug"
"strings"
"time"
"github.com/dustin/go-humanize"
@@ -52,7 +53,7 @@ type ServiceConfig struct {
// Allowlist is the list of hosts that don't require signatures
Allowlist []string
// MaxConcurrentProcessing is the most images processed at once; zero
// uses the image processor's default, half the CPUs
// uses the image processor's default, one per CPU
MaxConcurrentProcessing int
// Logger for logging
Logger *slog.Logger
@@ -128,26 +129,20 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
// ErrNegativeCached is returned when a URL is in the negative cache (recently failed).
var ErrNegativeCached = errors.New("request is in negative cache (recently failed)")
// ErrProcessingTimedOut is returned when processing an image stopped because
// the downstream_timeout of the request doing it ended, to that request and to
// every request that waited for its image.
var ErrProcessingTimedOut = errors.New(
"image processing stopped at the deadline of the request doing it")
// Get retrieves a processed image, fetching and processing if necessary.
// While its source URL is in the negative cache, a cached variant is still
// served and a variant not in the cache is still made from a cached source;
// only a request that needs the source fetched is refused.
func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, error) {
// Propagate AllowHTTP setting to the request
req.AllowHTTP = s.allowHTTP
// Check negative cache first - skip fetching for recently-failed URLs
negHit, err := s.cache.checkNegativeCache(ctx, req)
if err != nil {
s.log.Warn("negative cache check failed", "error", err)
}
if negHit {
s.log.Debug("negative cache hit",
"host", req.SourceHost,
"path", req.SourcePath,
)
return nil, fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
}
// Check variant cache first (disk only, no DB)
result, err := s.cache.Lookup(ctx, req)
if err != nil {
@@ -218,7 +213,16 @@ func (s *Service) ValidateRequest(req *ImageRequest) error {
return fmt.Errorf("invalid source URL: %w", err)
}
if s.allowlist.IsAllowed(parsedURL) {
// An allowlist entry names no port, so a host followed by ":", with or
// without a port, needs a signature. The port is read from the parsed
// URL, as the fetcher reads it: in an IPv6 address written without
// brackets, such as 2001:db8::1:8443, it is what follows the last ":".
// This is checked here and not in allowlist.IsAllowed, which also matches
// referer_blocklist, where a Referer must be blocked whatever its port.
hostHasPort := parsedURL.Port() != "" ||
strings.HasSuffix(parsedURL.Host, ":")
if !hostHasPort && s.allowlist.IsAllowed(parsedURL) {
return nil
}
@@ -423,12 +427,28 @@ func (s *Service) processFromSourceOrFetch(
// fetchAndProcess fetches from upstream, processes, and caches the result.
// It counts the fetch with the bytes read from upstream, including when
// reading the response or a later step fails.
// reading the response or a later step fails. It fetches nothing while the
// source URL is in the negative cache.
func (s *Service) fetchAndProcess(
ctx context.Context,
req *ImageRequest,
cacheKey VariantKey,
) (*processedVariant, error) {
// Check negative cache - skip fetching for recently-failed URLs
negHit, err := s.cache.checkNegativeCache(ctx, req)
if err != nil {
s.log.Warn("negative cache check failed", "error", err)
}
if negHit {
s.log.Debug("negative cache hit",
"host", req.SourceHost,
"path", req.SourcePath,
)
return nil, fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
}
// Fetch from upstream
sourceURL := req.SourceURL()
@@ -535,7 +555,13 @@ func (s *Service) processAndStore(
FitMode: imageprocessor.FitMode(req.FitMode),
}
// ctx has the deadline of the request doing the processing, which can
// end while a request waiting for its image has time left
processResult, err := s.processor.Process(ctx, source, processReq)
if errors.Is(err, context.DeadlineExceeded) {
return nil, fmt.Errorf("%w: %w", ErrProcessingTimedOut, err)
}
if err != nil {
return nil, fmt.Errorf("image processing failed: %w", err)
}
@@ -250,6 +250,97 @@ func TestService_ValidateRequest_SignatureExactHostMatch(t *testing.T) {
}
}
// TestService_ValidateRequest_HostWithPortNeedsSignature verifies that a host
// followed by a port, or by a colon with no port, matches no allowlist entry,
// so that a request for it without a signature is refused, while the same
// host with no port is allowlisted.
func TestService_ValidateRequest_HostWithPortNeedsSignature(t *testing.T) {
t.Parallel()
svc, _ := SetupTestService(t,
WithSigningKey(testSigningKey),
WithAllowlist(testHostCDN, "203.0.113.5", "2001:db8::1"),
)
tests := []struct {
host string
allowlisted bool
}{
{testHostCDN, true},
{"cdn.example.com:8443", false},
{"cdn.example.com:443", false},
{"cdn.example.com:", false},
{"203.0.113.5", true},
{"203.0.113.5:8080", false},
{"[2001:db8::1]", true},
{"[2001:db8::1]:8443", false},
}
for _, tt := range tests {
t.Run(tt.host, func(t *testing.T) {
t.Parallel()
req := &ImageRequest{
SourceHost: tt.host,
SourcePath: testPathCat,
Size: Size{Width: 50, Height: 50},
Format: FormatJPEG,
Quality: 85,
FitMode: FitCover,
}
err := svc.ValidateRequest(req)
t.Logf("ValidateRequest() for host %q with no signature: %v",
tt.host, err)
if allowlisted := err == nil; allowlisted != tt.allowlisted {
t.Errorf("host %q served without a signature = %v, want %v",
tt.host, allowlisted, tt.allowlisted)
}
})
}
}
// TestService_ValidateRequest_UnbracketedIPv6PortNeedsSignature verifies that
// an allowlisted IPv6 address written without brackets and followed by a
// port, or by a colon with no port, needs a signature. Go's URL parser, which
// the fetcher uses, reads 2001:db8::1:8443 as the host 2001:db8::1 with the
// port 8443.
func TestService_ValidateRequest_UnbracketedIPv6PortNeedsSignature(
t *testing.T,
) {
t.Parallel()
svc, _ := SetupTestService(t,
WithSigningKey(testSigningKey),
WithAllowlist("2001:db8::1"),
)
for _, host := range []string{"2001:db8::1:8443", "2001:db8::1:"} {
t.Run(host, func(t *testing.T) {
t.Parallel()
req := &ImageRequest{
SourceHost: host,
SourcePath: testPathCat,
Size: Size{Width: 50, Height: 50},
Format: FormatJPEG,
Quality: 85,
FitMode: FitCover,
}
err := svc.ValidateRequest(req)
t.Logf("ValidateRequest() for host %q with no signature: %v",
host, err)
if err == nil {
t.Errorf("host %q served without a signature, want refused",
host)
}
})
}
}
func TestService_Get_InvalidFile(t *testing.T) {
t.Parallel()
+12
View File
@@ -15,6 +15,7 @@ import (
var (
ErrInvalidPath = errors.New("invalid image path")
ErrMissingHost = errors.New("missing source host")
ErrInvalidHost = errors.New("invalid source host")
ErrMissingSize = errors.New("missing size specification")
ErrInvalidSize = errors.New("invalid size format")
ErrInvalidFormat = errors.New("invalid or unsupported format")
@@ -143,6 +144,17 @@ func parseImageComponents(remainder string) (*ParsedURL, error) {
return nil, ErrMissingHost
}
// A URL reads what comes before "@" as a user name and password, and the
// rest as the host
if strings.Contains(host, "@") {
return nil, fmt.Errorf("%w: %q holds @", ErrInvalidHost, host)
}
// Host names do not depend on letter case, so the allowlist, the upstream
// connection limits, the cache and the signature see one spelling of each
// host
host = strings.ToLower(host)
// Extract query string if present in path
if qIndex := strings.Index(path, "?"); qIndex != -1 {
query = path[qIndex+1:]
@@ -234,6 +234,49 @@ func TestParseImagePath(t *testing.T) {
}
}
// TestParseImagePath_LowerCasesHost verifies that the host is lower-cased, so
// that every spelling of its letter case is the same host to the allowlist,
// the upstream connection limits and the cache, and that the path keeps its
// case.
func TestParseImagePath_LowerCasesHost(t *testing.T) {
t.Parallel()
got, err := ParseImagePath("CDN.Example.COM/Photos/Cat.jpg/800x600.webp")
if err != nil {
t.Fatalf("ParseImagePath() error = %v", err)
}
t.Logf("parsed host %q, path %q", got.Host, got.Path)
if got.Host != testHostCDN {
t.Errorf("Host = %q, want %q", got.Host, testHostCDN)
}
if got.Path != "/Photos/Cat.jpg" {
t.Errorf("Path = %q, want %q", got.Path, "/Photos/Cat.jpg")
}
}
// TestParseImagePath_RefusesUserinfo verifies that a host holding "@" is
// refused: a URL reads what comes before the "@" as a user name and password
// and the rest as the host.
func TestParseImagePath_RefusesUserinfo(t *testing.T) {
t.Parallel()
for _, input := range []string{
"user@cdn.example.com/photos/cat.jpg/800x600.webp",
"user:password@cdn.example.com/photos/cat.jpg/800x600.webp",
} {
got, err := ParseImagePath(input)
t.Logf("ParseImagePath(%q) = %+v, error = %v", input, got, err)
if !errors.Is(err, ErrInvalidHost) {
t.Errorf("ParseImagePath(%q) error = %v, want %v",
input, err, ErrInvalidHost)
}
}
}
func TestParsedURL_ToImageRequest(t *testing.T) {
t.Parallel()