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..3e34e9e 100644 --- a/internal/imageprocessor/deadline_internal_test.go +++ b/internal/imageprocessor/deadline_internal_test.go @@ -88,3 +88,31 @@ func checkProcessStopsAtDeadline( t.Errorf("%d slots still held after Process() returned", held) } } + +// 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 +} + +// TestProcessReadsNoInputOnceDeadlinePassed checks that Process, called with +// a context whose deadline passed before it starts, gives its slot back +// without reading its 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. +func TestProcessReadsNoInputOnceDeadlinePassed(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithDeadline(t.Context(), + time.Now().Add(-time.Second)) + defer cancel() + + for range 100 { + checkProcessStopsAtDeadline(ctx, t, inputNotToBeRead{t: 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 52684a9..297116d 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -128,6 +128,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. func (s *Service) Get(ctx context.Context, req *ImageRequest) (*ImageResponse, error) { // Propagate AllowHTTP setting to the request