Bound concurrent image processing and upstream fetches (closes #64)

Nothing bounded total in-flight work, so a burst of cache misses across
hosts could exhaust memory. Two settings now do: max_concurrent_processing
(default the CPUs Go uses) and upstream_connections (default 64, beside
the per-host limit). A request that finds either full waits up to 10
seconds, then gets 503; a slot is released on every path. No request
holds source bytes while it waits: a cached source is read only after
the processing slot is taken, and a fetched one only while it holds its
upstream connection. libvips runs one worker thread per image with its
operation cache off. Both waits count toward downstream_timeout, as the
README says.

Model: opus-5-5

internal/imgcache/cache.go

@@ -383,13 +383,16 @@ func (c *Cache) GetSourceMetadataID(
 	return id, nil
 }
 
-// GetSourceContent returns a reader for cached source content by its hash.
-func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
+// GetSourceContent returns a reader for cached source content by its hash,
+// and the content's size in bytes.
+func (c *Cache) GetSourceContent(
+	contentHash ContentHash,
+) (io.ReadCloser, int64, error) {
 	if c.disabled {
-		return nil, ErrNotFound
+		return nil, 0, ErrNotFound
 	}
 
-	return c.srcContent.Load(contentHash)
+	return c.srcContent.LoadWithSize(contentHash)
 }
 
 // CleanExpired removes expired entries from the cache.

internal/imgcache/max_concurrent_processing_internal_test.go

@@ -0,0 +1,125 @@
+package imgcache
+
+import (
+	"image/color"
+	"image/jpeg"
+	"io"
+	"os"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/pixa/internal/imageprocessor"
+)
+
+// widthOnlyRequest asks for the test photo at width, its height scaled to
+// keep the photo's aspect ratio.
+func widthOnlyRequest(fixtures *TestFixtures, width int) *ImageRequest {
+	return &ImageRequest{
+		SourceHost: fixtures.GoodHost,
+		SourcePath: testPathPhoto,
+		Size:       Size{Width: width},
+		Format:     FormatJPEG,
+		Quality:    85,
+		FitMode:    FitCover,
+	}
+}
+
+// holdProcessingSlot takes one of proc's processing slots and returns the
+// func that gives it back. Process takes its slot before it reads its input,
+// so once it has read a byte from the pipe it holds the slot, until the pipe
+// is closed.
+func holdProcessingSlot(
+	t *testing.T, proc *imageprocessor.ImageProcessor,
+) func() {
+	t.Helper()
+
+	input, feed := io.Pipe()
+
+	go func() {
+		_, _ = proc.Process(t.Context(), input, &imageprocessor.Request{})
+	}()
+
+	_, err := feed.Write([]byte{0})
+	if err != nil {
+		t.Fatalf("Process call to hold the slot did not start: %v", err)
+	}
+
+	release := func() { _ = feed.Close() }
+	t.Cleanup(release)
+
+	return release
+}
+
+// TestService_Get_WaitsForSlotBeforeReadingCachedSource checks that a
+// request whose source is cached holds none of it while it waits for a
+// processing slot: it reads the cached file only once it has a slot. With
+// the only slot held, a request for a new width of the cached 100x100 photo
+// waits; the cached file is then rewritten as a 100x50 image before the slot
+// is freed, so the request must answer with that image scaled to 40x20.
+func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
+	t.Parallel()
+
+	svc, fixtures := SetupTestService(t)
+	svc.processor = imageprocessor.New(
+		imageprocessor.Params{MaxConcurrentProcessing: 1},
+	)
+
+	// A first request caches the photo as a source.
+	resp, err := svc.Get(t.Context(), widthOnlyRequest(fixtures, 50))
+	if err != nil {
+		t.Fatalf("first Get() error = %v", err)
+	}
+
+	_ = resp.Content.Close()
+
+	contentHash, _, err := svc.cache.LookupSource(t.Context(),
+		widthOnlyRequest(fixtures, 50))
+	if err != nil || contentHash == "" {
+		t.Fatalf("LookupSource() = %q, %v; want the cached source",
+			contentHash, err)
+	}
+
+	release := holdProcessingSlot(t, svc.processor)
+
+	var (
+		waited    *ImageResponse
+		waitedErr error
+	)
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		waited, waitedErr = svc.Get(t.Context(), widthOnlyRequest(fixtures, 40))
+	}()
+
+	// Give the request time to reach the slot: had it read the cached source
+	// before waiting, it would have read it by now.
+	time.Sleep(100 * time.Millisecond)
+
+	err = os.WriteFile(svc.cache.srcContent.hashToPath(contentHash),
+		generateTestJPEG(t, 100, 50, color.RGBA{0, 0, 255, 255}), 0o600)
+	if err != nil {
+		t.Fatalf("failed to rewrite the cached source: %v", err)
+	}
+
+	release()
+	<-done
+
+	if waitedErr != nil {
+		t.Fatalf("Get() error = %v", waitedErr)
+	}
+
+	defer func() { _ = waited.Content.Close() }()
+
+	output, err := jpeg.DecodeConfig(waited.Content)
+	if err != nil {
+		t.Fatalf("failed to decode the response: %v", err)
+	}
+
+	if output.Width != 40 || output.Height != 20 {
+		t.Errorf("response is %dx%d, want 40x20: the request read the cached "+
+			"source before it had a processing slot", output.Width, output.Height)
+	}
+}

internal/imgcache/service.go

@@ -43,6 +43,9 @@ type ServiceConfig struct {
 	SigningKey string
 	// Allowlist is the list of hosts that don't require signatures
 	Allowlist []string
+	// MaxConcurrentProcessing is the most images processed at once; zero
+	// uses the image processor's default, one per CPU
+	MaxConcurrentProcessing int
 	// Logger for logging
 	Logger *slog.Logger
 }
@@ -91,9 +94,10 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
 	}
 
 	maxResponseSize := fetcherCfg.MaxResponseSize
-	processor := imageprocessor.New(
-		imageprocessor.Params{MaxInputBytes: maxResponseSize},
-	)
+	processor := imageprocessor.New(imageprocessor.Params{
+		MaxInputBytes:           maxResponseSize,
+		MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
+	})
 
 	return &Service{
 		cache:           cfg.Cache,
@@ -237,38 +241,37 @@ func (s *Service) GenerateSignedURL(
 		baseURL, path, sig, exp, req.Quality, req.FitMode), nil
 }
 
-// loadCachedSource attempts to load source content from cache, returning nil
-// if the cached data is unavailable or exceeds maxResponseSize.
-func (s *Service) loadCachedSource(contentHash ContentHash) []byte {
-	reader, err := s.cache.GetSourceContent(contentHash)
+// loadCachedSource opens source content from cache, without reading it, and
+// returns it with its size; nil if the cached data is unavailable, empty or
+// exceeds maxResponseSize.
+func (s *Service) loadCachedSource(
+	contentHash ContentHash,
+) (io.ReadCloser, int64) {
+	reader, size, err := s.cache.GetSourceContent(contentHash)
 	if err != nil {
 		s.log.Warn("failed to load cached source, fetching", "error", err)
 
-		return nil
+		return nil, 0
 	}
 
-	// Bound the read to maxResponseSize to prevent unbounded memory use
-	// from unexpectedly large cached files.
-	limited := io.LimitReader(reader, s.maxResponseSize+1)
-	data, err := io.ReadAll(limited)
-	_ = reader.Close()
+	if size > s.maxResponseSize {
+		_ = reader.Close()
 
-	if err != nil {
-		s.log.Warn("failed to read cached source, fetching", "error", err)
-
-		return nil
-	}
-
-	if int64(len(data)) > s.maxResponseSize {
 		s.log.Warn("cached source exceeds max response size, discarding",
 			"hash", contentHash,
 			"max_bytes", s.maxResponseSize,
 		)
 
-		return nil
+		return nil, 0
 	}
 
-	return data
+	if size == 0 {
+		_ = reader.Close()
+
+		return nil, 0
+	}
+
+	return reader, size
 }
 
 // processFromSourceOrFetch processes an image, using cached source content
@@ -285,22 +288,27 @@ func (s *Service) processFromSourceOrFetch(
 		s.log.Warn("source lookup failed", "error", err)
 	}
 
-	var sourceData []byte
+	var (
+		source     io.ReadCloser
+		sourceSize int64
+	)
 
 	if contentHash != "" {
 		s.log.Debug("using cached source", "hash", contentHash)
-		sourceData = s.loadCachedSource(contentHash)
+		source, sourceSize = s.loadCachedSource(contentHash)
 	}
 
 	// Fetch from upstream if we don't have source data or it's empty
-	if len(sourceData) == 0 {
+	if source == nil {
 		return s.fetchAndProcess(ctx, req, cacheKey)
 	}
 
-	// Process using cached source; nothing was fetched from upstream
-	resp, err := s.processAndStore(
-		ctx, req, cacheKey, sourceData, int64(len(sourceData)),
-	)
+	defer func() { _ = source.Close() }()
+
+	// Process using cached source; nothing was fetched from upstream. The
+	// image processor reads the source only once it has a processing slot,
+	// so a request waiting for one holds none of it in memory.
+	resp, err := s.processAndStore(ctx, req, cacheKey, source, sourceSize)
 
 	return resp, 0, err
 }
@@ -334,6 +342,10 @@ func (s *Service) fetchAndProcess(
 		return nil, 0, fmt.Errorf("upstream fetch failed: %w", err)
 	}
 
+	// Closing the body frees the upstream connection. It is closed only
+	// after processing, so the fetcher's connection limit also bounds the
+	// fetched sources held in memory while their requests wait for a
+	// processing slot.
 	defer func() { _ = fetchResult.Content.Close() }()
 
 	// Read and validate the source content
@@ -379,17 +391,20 @@ func (s *Service) fetchAndProcess(
 		// Continue even if caching fails
 	}
 
-	resp, err := s.processAndStore(ctx, req, cacheKey, sourceData, fetchBytes)
+	resp, err := s.processAndStore(
+		ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
+	)
 
 	return resp, fetchBytes, err
 }
 
-// processAndStore processes an image and stores the result.
+// processAndStore processes the image read from source and stores the
+// result.
 func (s *Service) processAndStore(
 	ctx context.Context,
 	req *ImageRequest,
 	cacheKey VariantKey,
-	sourceData []byte,
+	source io.Reader,
 	fetchBytes int64,
 ) (*ImageResponse, error) {
 	// Process the image
@@ -402,7 +417,7 @@ func (s *Service) processAndStore(
 		FitMode: imageprocessor.FitMode(req.FitMode),
 	}
 
-	processResult, err := s.processor.Process(ctx, bytes.NewReader(sourceData), processReq)
+	processResult, err := s.processor.Process(ctx, source, processReq)
 	if err != nil {
 		return nil, fmt.Errorf("image processing failed: %w", err)
 	}