Stop a transcode whose request has ended before it encodes (closes #243)

Process looked at its context only while waiting for a processing
slot, so a transcode kept its slot and worked on after its request's
downstream_timeout. acquireSlot now gives the slot back when the
context has ended by the time it has one, and encode returns the
context's error without beginning. govips gives pixa no way to stop
libvips once an encode has begun, so that encode keeps its slot. A
request waiting on a shared transcode stopped at the first request's
deadline is answered 503, not 500. README.md "Routes" says both.

Model: opus-5-5

internal/handlers/image.go

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

internal/handlers/imageenc.go

@@ -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),

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

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

internal/imageprocessor/imageprocessor.go

@@ -208,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,
@@ -290,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)
 	}
@@ -417,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.
@@ -527,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
 	}
@@ -540,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)
 		}
@@ -632,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.

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;
@@ -549,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)
 	}