Compare commits
3
Commits
1ae38a2425
...
822f03193a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
822f03193a | ||
|
|
d5052495c7 | ||
|
|
6572b3fe08 |
@@ -319,9 +319,12 @@ fetch sends the first request's ID upstream, and the lines logged for the fetch
|
|||||||
and the transcode carry that ID.
|
and the transcode carry that ID.
|
||||||
|
|
||||||
A transcode whose request's `downstream_timeout` has ended (for a shared one,
|
A transcode whose request's `downstream_timeout` has ended (for a shared one,
|
||||||
the first request's) stops when it gets its processing slot or before it begins
|
the first request's) stops waiting for a processing slot, or, once it has one,
|
||||||
encoding, and gives the slot back. libvips cannot stop an encode once it has
|
gives it back without beginning to encode. A request waiting for a shared
|
||||||
begun, so an encode keeps its slot until it finishes, which can be after
|
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
|
`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,
|
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,
|
encoding 8192x8192 random pixels takes up to about 1 second as JPEG, 15 as PNG,
|
||||||
|
|||||||
@@ -30,6 +30,24 @@ P2: security: per-origin rate limiting
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 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
|
- 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 (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
|
cache, so once a fetch of the source failed, as one does when another variant
|
||||||
@@ -59,21 +77,6 @@ P2: security: per-origin rate limiting
|
|||||||
and the query now have each `%` written as `%25` and each `:` as `%3A`; a path
|
and the query now have each `%` written as `%25` and each `:` as `%3A`; a path
|
||||||
and query holding neither are signed and keyed as before. Signature
|
and query holding neither are signed and keyed as before. Signature
|
||||||
Specification in `README.md` says so.
|
Specification in `README.md` says so.
|
||||||
- 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. libvips cannot stop an encode once it 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 the example config allowlists no host (closes #249): it allowlisted
|
- 2026-10-15 the example config allowlists no host (closes #249): it allowlisted
|
||||||
`github.com` and `user-images.githubusercontent.com`, where anyone can put an
|
`github.com` and `user-images.githubusercontent.com`, where anyone can put an
|
||||||
image, so a deployment set up as Getting Started says served those images to
|
image, so a deployment set up as Getting Started says served those images to
|
||||||
|
|||||||
@@ -240,7 +240,8 @@ func (s *Handlers) respondImageError(
|
|||||||
}
|
}
|
||||||
|
|
||||||
if errors.Is(err, httpfetcher.ErrTooManyConnections) ||
|
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",
|
s.respondError(w, "server busy, try again later",
|
||||||
http.StatusServiceUnavailable)
|
http.StatusServiceUnavailable)
|
||||||
|
|
||||||
|
|||||||
@@ -158,7 +158,8 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
|
|||||||
case errors.Is(err, httpfetcher.ErrUpstreamTimeout):
|
case errors.Is(err, httpfetcher.ErrUpstreamTimeout):
|
||||||
s.respondError(w, "upstream timeout", http.StatusGatewayTimeout)
|
s.respondError(w, "upstream timeout", http.StatusGatewayTimeout)
|
||||||
case errors.Is(err, httpfetcher.ErrTooManyConnections),
|
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",
|
s.respondError(w, "server busy, try again later",
|
||||||
http.StatusServiceUnavailable)
|
http.StatusServiceUnavailable)
|
||||||
case errors.Is(err, httpfetcher.ErrInvalidContentType),
|
case errors.Is(err, httpfetcher.ErrInvalidContentType),
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package handlers
|
package handlers
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -47,3 +48,39 @@ func TestServerBusyAnswers503(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestProcessingTimedOutAnswers503 checks that both 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.
|
||||||
|
func TestProcessingTimedOutAnswers503(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
h := &Handlers{log: slog.New(slog.DiscardHandler)}
|
||||||
|
req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.jpg"}
|
||||||
|
err := fmt.Errorf("%w: failed to encode: %w",
|
||||||
|
imgcache.ErrProcessingTimedOut, context.DeadlineExceeded)
|
||||||
|
|
||||||
|
plain := httptest.NewRecorder()
|
||||||
|
h.respondImageError(plain, req, err)
|
||||||
|
|
||||||
|
encrypted := httptest.NewRecorder()
|
||||||
|
h.handleImageError(encrypted, err)
|
||||||
|
|
||||||
|
for route, rec := range map[string]*httptest.ResponseRecorder{
|
||||||
|
"/v1/image/": plain, "/v1/e/": encrypted,
|
||||||
|
} {
|
||||||
|
t.Logf("%s: status = %d, body = %q", route, rec.Code, rec.Body.String())
|
||||||
|
|
||||||
|
if rec.Code != http.StatusServiceUnavailable {
|
||||||
|
t.Errorf("%s: status = %d, want %d",
|
||||||
|
route, 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",
|
||||||
|
route, rec.Body.String())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -28,11 +28,23 @@ func (r *readerAfterDeadline) Read(p []byte) (int, error) {
|
|||||||
return r.data.Read(p)
|
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
|
// TestProcessStopsWhenDeadlinePassed checks that Process returns
|
||||||
// context.DeadlineExceeded, and holds no slot afterwards, when its context's
|
// context.DeadlineExceeded, and holds no slot afterwards, when its context's
|
||||||
// deadline has passed before it starts or while it reads its input. libvips
|
// deadline has passed before it starts, then without reading its input, or
|
||||||
// cannot stop an encode once it has begun, so Process must not begin one for
|
// while it reads its input. govips gives pixa no way to stop libvips once an
|
||||||
// a request that has ended.
|
// encode has begun, so Process must not begin one for a request that has
|
||||||
|
// ended.
|
||||||
func TestProcessStopsWhenDeadlinePassed(t *testing.T) {
|
func TestProcessStopsWhenDeadlinePassed(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -45,7 +57,11 @@ func TestProcessStopsWhenDeadlinePassed(t *testing.T) {
|
|||||||
time.Now().Add(-time.Second))
|
time.Now().Add(-time.Second))
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
checkProcessStopsAtDeadline(ctx, t, bytes.NewReader(input))
|
// 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.Run("while Process reads its input", func(t *testing.T) {
|
||||||
|
|||||||
@@ -209,8 +209,9 @@ func New(params Params) *ImageProcessor {
|
|||||||
// MaxConcurrentProcessing images are already being processed, it waits up
|
// MaxConcurrentProcessing images are already being processed, it waits up
|
||||||
// to ProcessingWaitTimeout for one to finish, then fails with
|
// to ProcessingWaitTimeout for one to finish, then fails with
|
||||||
// ErrTooManyImages. When ctx has ended once Process has a slot, or before it
|
// ErrTooManyImages. When ctx has ended once Process has a slot, or before it
|
||||||
// encodes, it returns ctx's error and gives the slot back. libvips cannot stop
|
// encodes, it returns ctx's error and gives the slot back. govips gives pixa
|
||||||
// an encode once it has begun, so an encode keeps its slot until it finishes.
|
// no way to stop libvips once an encode has begun, so an encode keeps its
|
||||||
|
// slot until it finishes.
|
||||||
func (p *ImageProcessor) Process(
|
func (p *ImageProcessor) Process(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
input io.Reader,
|
input io.Reader,
|
||||||
@@ -532,9 +533,9 @@ func (p *ImageProcessor) resize(
|
|||||||
|
|
||||||
const defaultQuality = 85
|
const defaultQuality = 85
|
||||||
|
|
||||||
// encode encodes an image to the specified format. libvips cannot stop an
|
// encode encodes an image to the specified format. govips gives pixa no way
|
||||||
// encode once it has begun, so when ctx has ended encode returns ctx's error
|
// to stop libvips once an encode has begun, so when ctx has ended encode
|
||||||
// without beginning one.
|
// returns ctx's error without beginning one.
|
||||||
func (p *ImageProcessor) encode(
|
func (p *ImageProcessor) encode(
|
||||||
ctx context.Context, img *vips.ImageRef, format Format, quality int,
|
ctx context.Context, img *vips.ImageRef, format Format, quality int,
|
||||||
) ([]byte, error) {
|
) ([]byte, error) {
|
||||||
@@ -644,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
|
// 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
|
// 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,
|
// seconds to save an 8192x8192 RGB image of random pixels, and about 110
|
||||||
// against the default downstream_timeout of 60 seconds. On an image of
|
// with an alpha channel, the worst case, against the default
|
||||||
// milder noise, which 1 saves in about 12 seconds, 2 takes nearly a minute
|
// downstream_timeout of 60 seconds. On an image of milder noise, which 1
|
||||||
// and libvips' default, 4, takes minutes.
|
// saves in about 12 seconds, 2 takes nearly a minute and libvips' default,
|
||||||
|
// 4, takes minutes.
|
||||||
const avifEffort = 1
|
const avifEffort = 1
|
||||||
|
|
||||||
// avifBitdepth is the AVIF bit depth, 8 bits per sample for every image.
|
// avifBitdepth is the AVIF bit depth, 8 bits per sample for every image.
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import (
|
|||||||
|
|
||||||
"github.com/getsentry/sentry-go"
|
"github.com/getsentry/sentry-go"
|
||||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||||
|
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||||
"sneak.berlin/go/pixa/internal/magic"
|
"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.
|
// panickingFetcher panics on every fetch.
|
||||||
type panickingFetcher struct{}
|
type panickingFetcher struct{}
|
||||||
|
|
||||||
|
|||||||
@@ -129,6 +129,12 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
|
|||||||
// ErrNegativeCached is returned when a URL is in the negative cache (recently failed).
|
// ErrNegativeCached is returned when a URL is in the negative cache (recently failed).
|
||||||
var ErrNegativeCached = errors.New("request is in 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.
|
// Get retrieves a processed image, fetching and processing if necessary.
|
||||||
// While its source URL is in the negative cache, a cached variant is still
|
// 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;
|
// served and a variant not in the cache is still made from a cached source;
|
||||||
@@ -549,7 +555,13 @@ func (s *Service) processAndStore(
|
|||||||
FitMode: imageprocessor.FitMode(req.FitMode),
|
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)
|
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 {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("image processing failed: %w", err)
|
return nil, fmt.Errorf("image processing failed: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user