diff --git a/internal/handlers/server_busy_internal_test.go b/internal/handlers/server_busy_internal_test.go index 2f3471c..25a4d8f 100644 --- a/internal/handlers/server_busy_internal_test.go +++ b/internal/handlers/server_busy_internal_test.go @@ -1,6 +1,7 @@ package handlers import ( + "context" "fmt" "log/slog" "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()) + } + } +} diff --git a/internal/imageprocessor/deadline_internal_test.go b/internal/imageprocessor/deadline_internal_test.go index 4d35ad0..1bde965 100644 --- a/internal/imageprocessor/deadline_internal_test.go +++ b/internal/imageprocessor/deadline_internal_test.go @@ -28,11 +28,23 @@ func (r *readerAfterDeadline) Read(p []byte) (int, error) { 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 or while it reads its input. libvips -// cannot stop an encode once it has begun, so Process must not begin one for -// a request that has ended. +// 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() @@ -45,7 +57,11 @@ func TestProcessStopsWhenDeadlinePassed(t *testing.T) { time.Now().Add(-time.Second)) 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) { diff --git a/internal/imgcache/concurrent_misses_internal_test.go b/internal/imgcache/concurrent_misses_internal_test.go index ad0e1b2..b81d63a 100644 --- a/internal/imgcache/concurrent_misses_internal_test.go +++ b/internal/imgcache/concurrent_misses_internal_test.go @@ -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{} diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 57f4e82..2ba0e7f 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -129,6 +129,12 @@ 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;