internal/handlers/server_busy_internal_test.go

@@ -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())
+		}
+	}
+}

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) {

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{}
 

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;