Test the processing and upstream connection limits (closes #64)

Failing tests for two limits that do not exist yet. Config:
max_concurrent_processing and upstream_connections, their defaults, and
valid and invalid values from the file and the environment. Image
processor: never more images at once than its limit, waiting and then
failing with ErrTooManyImages when no slot frees, and freeing its slot on
every error. Fetcher: connections to all hosts counted together, apart
from the per-host limit, and freed on errors. Both image routes answer
503 when either wait gives up. TestEnvironmentSetsEveryKey sets the two
new variables, as it compares the whole config. The tests do not compile
until the limits exist.

Model: opus-5-5

internal/config/concurrency_limits_internal_test.go

@@ -0,0 +1,162 @@
+package config
+
+import (
+	"runtime"
+	"testing"
+)
+
+// The variables that set the two concurrency limits.
+const (
+	testMaxConcurrentProcessingVar = "PIXA_MAX_CONCURRENT_PROCESSING"
+	testUpstreamConnectionsVar     = "PIXA_UPSTREAM_CONNECTIONS"
+)
+
+// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
+// max_concurrent_processing is the number of CPUs Go uses and an omitted
+// upstream_connections is 64.
+func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine)
+	if err != nil {
+		t.Fatalf("minimal config should be valid, got error: %v", err)
+	}
+
+	if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
+		t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
+			c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
+	}
+
+	if c.UpstreamConnections != 64 {
+		t.Errorf("UpstreamConnections = %d, want 64", c.UpstreamConnections)
+	}
+}
+
+// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
+// two limits are used as given.
+func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
+	t.Parallel()
+
+	c, err := configFromYAML(t, signingKeyLine+
+		"max_concurrent_processing: 3\nupstream_connections: 10\n")
+	if err != nil {
+		t.Fatalf("valid config should load, got error: %v", err)
+	}
+
+	if c.MaxConcurrentProcessing != 3 {
+		t.Errorf("MaxConcurrentProcessing = %d, want 3", c.MaxConcurrentProcessing)
+	}
+
+	if c.UpstreamConnections != 10 {
+		t.Errorf("UpstreamConnections = %d, want 10", c.UpstreamConnections)
+	}
+}
+
+// TestInvalidConcurrencyLimitAbortsStartup checks that a limit that is
+// not a whole number of at least 1, or is null, aborts startup naming the
+// key and the value, and the variable too where the value could have come
+// from it.
+func TestInvalidConcurrencyLimitAbortsStartup(t *testing.T) {
+	t.Parallel()
+
+	processing := keyMaxConcurrentProcessing
+	connections := keyUpstreamConnections
+
+	runAbortCases(t, []abortCase{
+		{
+			name: "max_concurrent_processing zero",
+			yaml: signingKeyLine + processing + ": 0\n",
+			wantErrSubstrings: []string{
+				processing, testMaxConcurrentProcessingVar, "value 0",
+			},
+		},
+		{
+			name: "max_concurrent_processing negative",
+			yaml: signingKeyLine + processing + ": -2\n",
+			wantErrSubstrings: []string{
+				processing, testMaxConcurrentProcessingVar, "value -2",
+			},
+		},
+		{
+			name: "max_concurrent_processing not a number",
+			yaml: signingKeyLine + processing + ": lots\n",
+			wantErrSubstrings: []string{
+				processing, testMaxConcurrentProcessingVar, "lots",
+			},
+		},
+		{
+			name:              "max_concurrent_processing fractional",
+			yaml:              signingKeyLine + processing + ": 1.5\n",
+			wantErrSubstrings: []string{processing, "1.5"},
+		},
+		{
+			name:              "max_concurrent_processing null",
+			yaml:              signingKeyLine + processing + ": null\n",
+			wantErrSubstrings: []string{processing, nullValueText},
+		},
+		{
+			name: "upstream_connections zero",
+			yaml: signingKeyLine + connections + ": 0\n",
+			wantErrSubstrings: []string{
+				connections, testUpstreamConnectionsVar, "value 0",
+			},
+		},
+		{
+			name: "upstream_connections negative",
+			yaml: signingKeyLine + connections + ": -5\n",
+			wantErrSubstrings: []string{
+				connections, testUpstreamConnectionsVar, "value -5",
+			},
+		},
+		{
+			name: "upstream_connections not a number",
+			yaml: signingKeyLine + connections + ": many\n",
+			wantErrSubstrings: []string{
+				connections, testUpstreamConnectionsVar, "many",
+			},
+		},
+		{
+			name:              "upstream_connections null",
+			yaml:              signingKeyLine + connections + ": null\n",
+			wantErrSubstrings: []string{connections, nullValueText},
+		},
+	})
+}
+
+// TestConcurrencyLimitsFromEnvironment checks that the two variables set
+// the limits over the config file, and that an invalid value in either
+// aborts startup naming the variable and the value.
+func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
+	t.Setenv(testMaxConcurrentProcessingVar, "3")
+	t.Setenv(testUpstreamConnectionsVar, "10")
+
+	c, err := configFromYAML(t, signingKeyLine+
+		"max_concurrent_processing: 5\nupstream_connections: 50\n")
+	if err != nil {
+		t.Fatalf("limits from the environment should load: %v", err)
+	}
+
+	if c.MaxConcurrentProcessing != 3 || c.UpstreamConnections != 10 {
+		t.Errorf("limits = %d and %d, want 3 and 10 from the environment",
+			c.MaxConcurrentProcessing, c.UpstreamConnections)
+	}
+
+	cases := []struct {
+		variable string
+		value    string
+	}{
+		{testMaxConcurrentProcessingVar, "lots"},
+		{testMaxConcurrentProcessingVar, "0"},
+		{testUpstreamConnectionsVar, "-1"},
+		{testUpstreamConnectionsVar, "ten"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.variable+"="+tc.value, func(t *testing.T) {
+			t.Setenv(tc.variable, tc.value)
+
+			_, err := configFromYAML(t, signingKeyLine)
+			wantStartupError(t, err, tc.variable, tc.value)
+		})
+	}
+}

internal/config/env_internal_test.go

@@ -66,6 +66,8 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
 	t.Setenv("PIXA_ALLOWLIST_HOSTS", "s3.sneak.cloud,.example.com")
 	t.Setenv("PIXA_ALLOW_HTTP", "true")
 	t.Setenv("PIXA_UPSTREAM_CONNECTIONS_PER_HOST", "5")
+	t.Setenv("PIXA_UPSTREAM_CONNECTIONS", "10")
+	t.Setenv("PIXA_MAX_CONCURRENT_PROCESSING", "3")
 	t.Setenv("PIXA_CACHE_MAX_BYTES", "1024")
 	t.Setenv("PIXA_BLOCKED_NETWORKS", "203.0.113.0/24")
 	t.Setenv("PIXA_TRUSTED_PROXIES", "192.0.2.0/24")
@@ -88,6 +90,8 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
 		AllowlistHosts:             []string{testHostS3, ".example.com"},
 		AllowHTTP:                  true,
 		UpstreamConnectionsPerHost: 5,
+		UpstreamConnections:        10,
+		MaxConcurrentProcessing:    3,
 		CacheMaxBytes:              1024,
 		cacheMaxBytesExplicit:      true,
 		BlockedNetworks:            []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")},

internal/handlers/server_busy_internal_test.go

@@ -0,0 +1,49 @@
+package handlers
+
+import (
+	"fmt"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"sneak.berlin/go/pixa/internal/imageprocessor"
+	"sneak.berlin/go/pixa/internal/imgcache"
+)
+
+// TestServerBusyAnswers503 checks that both image routes answer 503 with
+// a clear error when the image service gives up waiting for a free upstream
+// connection or processing slot, wrapped as the service wraps them.
+func TestServerBusyAnswers503(t *testing.T) {
+	t.Parallel()
+
+	h := &Handlers{log: slog.New(slog.DiscardHandler)}
+	req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.jpg"}
+
+	for _, err := range []error{
+		fmt.Errorf("upstream fetch failed: %w", httpfetcher.ErrTooManyConnections),
+		fmt.Errorf("image processing failed: %w", imageprocessor.ErrTooManyImages),
+	} {
+		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,
+		} {
+			if rec.Code != http.StatusServiceUnavailable {
+				t.Errorf("%s for %v: status = %d, want %d",
+					route, err, rec.Code, http.StatusServiceUnavailable)
+			}
+
+			if !strings.Contains(rec.Body.String(), "server busy, try again later") {
+				t.Errorf("%s for %v: body = %q, want the server busy error",
+					route, err, rec.Body.String())
+			}
+		}
+	}
+}

internal/httpfetcher/max_connections_internal_test.go

@@ -0,0 +1,128 @@
+package httpfetcher
+
+import (
+	"errors"
+	"net"
+	"strconv"
+	"testing"
+	"time"
+)
+
+// imageURLOnPort is the fake upstream's image route on testPublicHost at
+// port. Each port is a different host to the per-host limit, while the test
+// dialer sends every port to the one test server.
+func imageURLOnPort(port int) string {
+	return "http://" + net.JoinHostPort(testPublicHost, strconv.Itoa(port)) +
+		"/image"
+}
+
+func TestDefaultConfigMaxConnections(t *testing.T) {
+	t.Parallel()
+
+	if got := DefaultConfig().MaxConnections; got != DefaultMaxConnections {
+		t.Errorf("MaxConnections = %d, want %d", got, DefaultMaxConnections)
+	}
+}
+
+// TestFetchLimitsConnectionsToAllHostsTogether checks that MaxConnections
+// counts the fetches to every host together, apart from the per-host
+// limit: with MaxConnections at 2 and two responses open from two hosts, a
+// fetch from a third host, which has nothing open, waits the whole wait
+// timeout and fails with ErrTooManyConnections. Closing one response lets
+// it through.
+func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
+	t.Parallel()
+
+	srv := startUpstream(t)
+
+	cfg := DefaultConfig()
+	cfg.MaxConnections = 2
+
+	f, _ := newServerFetcher(t, srv, cfg)
+	f.connectionWaitTimeout = 100 * time.Millisecond
+
+	first, err := f.Fetch(testContext(t), imageURLOnPort(81))
+	if err != nil {
+		t.Fatalf("first Fetch() error = %v", err)
+	}
+
+	second, err := f.Fetch(testContext(t), imageURLOnPort(82))
+	if err != nil {
+		t.Fatalf("second Fetch() error = %v", err)
+	}
+
+	defer func() { _ = second.Content.Close() }()
+
+	start := time.Now()
+
+	_, err = f.Fetch(testContext(t), imageURLOnPort(83))
+	if !errors.Is(err, ErrTooManyConnections) {
+		t.Fatalf("third Fetch() error = %v, want ErrTooManyConnections", err)
+	}
+
+	if waited := time.Since(start); waited < f.connectionWaitTimeout {
+		t.Errorf("third Fetch() failed after %v, before waiting %v",
+			waited, f.connectionWaitTimeout)
+	}
+
+	if held := semLen(f, testPublicHost+":83"); held != 0 {
+		t.Errorf("the refused fetch kept its host's slot: %d held", held)
+	}
+
+	err = first.Content.Close()
+	if err != nil {
+		t.Fatalf("close first body: %v", err)
+	}
+
+	third, err := f.Fetch(testContext(t), imageURLOnPort(83))
+	if err != nil {
+		t.Fatalf("Fetch() after a response was closed: error = %v", err)
+	}
+
+	_ = third.Content.Close()
+}
+
+// TestFetchReleasesConnectionOnError checks that a fetch that fails after
+// taking its connection gives it back: with MaxConnections at 1, the slot
+// must be free after the failure and the next fetch must succeed.
+func TestFetchReleasesConnectionOnError(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		url  string
+		want error
+	}{
+		{"upstream answers 500", upstreamURL("/status/500"), ErrUpstreamError},
+		{"upstream sends HTML", upstreamURL("/html"), ErrInvalidContentType},
+		// 198.51.100.7 (TEST-NET-2) passes the SSRF checks, and the test
+		// dialer refuses every host but testPublicHost.
+		{"connecting fails", "http://198.51.100.7/image", errUnexpectedDial},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			srv := startUpstream(t)
+
+			cfg := DefaultConfig()
+			cfg.MaxConnections = 1
+
+			f, _ := newServerFetcher(t, srv, cfg)
+			f.connectionWaitTimeout = 100 * time.Millisecond
+
+			_, err := f.Fetch(testContext(t), tc.url)
+			if !errors.Is(err, tc.want) {
+				t.Fatalf("Fetch() error = %v, want %v", err, tc.want)
+			}
+
+			if held := len(f.allHostsSemaphore); held != 0 {
+				t.Fatalf("connection still held after the error: %d held", held)
+			}
+
+			res := fetchImage(t, f, "/image")
+			_ = res.Content.Close()
+		})
+	}
+}

internal/imageprocessor/max_concurrent_processing_internal_test.go

@@ -0,0 +1,302 @@
+package imageprocessor
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+	"testing/iotest"
+	"time"
+)
+
+// errTestReadFailed is the error the unreadable test input returns.
+var errTestReadFailed = errors.New("test input cannot be read")
+
+// readingCounter counts the Process calls reading their input at the same
+// time and remembers the most there ever were.
+type readingCounter struct {
+	mu      sync.Mutex
+	reading int
+	most    int
+}
+
+func (c *readingCounter) start() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.reading++
+	c.most = max(c.most, c.reading)
+}
+
+func (c *readingCounter) stop() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.reading--
+}
+
+func (c *readingCounter) mostReading() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.most
+}
+
+// gatedReader is a Process input. Its first Read counts the call in,
+// reports it on entered and blocks until gate is closed; it counts the call
+// out when it returns io.EOF. Process reads its input only while it holds a
+// processing slot, so the count never goes above MaxConcurrentProcessing.
+type gatedReader struct {
+	data    *bytes.Reader
+	gate    <-chan struct{}
+	entered chan<- struct{}
+	counter *readingCounter
+	started bool
+}
+
+func (r *gatedReader) Read(p []byte) (int, error) {
+	if !r.started {
+		r.started = true
+		r.counter.start()
+
+		r.entered <- struct{}{}
+
+		<-r.gate
+	}
+
+	n, err := r.data.Read(p)
+	if errors.Is(err, io.EOF) {
+		r.counter.stop()
+	}
+
+	return n, err
+}
+
+// smallJPEGRequest asks for a 5x5 JPEG.
+func smallJPEGRequest() *Request {
+	return &Request{
+		Size:    Size{Width: 5, Height: 5},
+		Format:  FormatJPEG,
+		Quality: 85,
+		FitMode: FitCover,
+	}
+}
+
+// processInBackground runs Process on reader in a new goroutine and sends
+// its error on results.
+func processInBackground(
+	proc *ImageProcessor, reader *gatedReader, results chan<- error,
+) {
+	go func() {
+		result, err := proc.Process(context.Background(), reader, smallJPEGRequest())
+		if err == nil {
+			_ = result.Content.Close()
+		}
+
+		results <- err
+	}()
+}
+
+// waitForEntries fails the test unless count Process calls report on
+// entered within a few seconds.
+func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
+	t.Helper()
+
+	for range count {
+		select {
+		case <-entered:
+		case <-time.After(5 * time.Second):
+			t.Fatal("Process calls did not start reading their input")
+		}
+	}
+}
+
+func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
+	t.Parallel()
+
+	for _, limit := range []int{0, -1} {
+		proc := New(Params{MaxConcurrentProcessing: limit})
+		if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
+			t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
+				limit, got, runtime.GOMAXPROCS(0))
+		}
+	}
+
+	proc := New(Params{MaxConcurrentProcessing: 3})
+	if got := cap(proc.processingSemaphore); got != 3 {
+		t.Errorf("MaxConcurrentProcessing 3: %d slots, want 3", got)
+	}
+}
+
+// TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
+// than MaxConcurrentProcessing allows and holds the first ones inside
+// Process until the test lets them go. No more than the limit may be
+// working at once, and the calls held back must wait for a slot and then
+// succeed.
+func TestProcessNeverExceedsMaxConcurrentProcessing(t *testing.T) {
+	t.Parallel()
+
+	const (
+		limit = 2
+		calls = 6
+	)
+
+	proc := New(Params{MaxConcurrentProcessing: limit})
+	input := createTestJPEG(t, 50, 50)
+
+	counter := &readingCounter{}
+	gate := make(chan struct{})
+	entered := make(chan struct{}, calls)
+	results := make(chan error, calls)
+
+	openGate := sync.OnceFunc(func() { close(gate) })
+	t.Cleanup(openGate)
+
+	for range calls {
+		processInBackground(proc, &gatedReader{
+			data: bytes.NewReader(input), gate: gate, entered: entered,
+			counter: counter,
+		}, results)
+	}
+
+	waitForEntries(t, entered, limit)
+
+	// A call beyond the limit would start reading its input now.
+	select {
+	case <-entered:
+		t.Fatalf("a Process call started while %d were already working", limit)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	openGate()
+
+	for range calls {
+		err := <-results
+		if err != nil {
+			t.Errorf("Process() error = %v, want nil once a slot is free", err)
+		}
+	}
+
+	if most := counter.mostReading(); most > limit {
+		t.Errorf("%d Process calls worked at once, want at most %d", most, limit)
+	}
+}
+
+// TestProcessWaitsThenFailsWhenNoSlotFrees holds the only slot and checks
+// that another call waits the whole wait timeout, then fails with
+// ErrTooManyImages instead of processing anyway.
+func TestProcessWaitsThenFailsWhenNoSlotFrees(t *testing.T) {
+	t.Parallel()
+
+	proc := New(Params{MaxConcurrentProcessing: 1})
+	proc.processingWaitTimeout = 100 * time.Millisecond
+
+	input := createTestJPEG(t, 10, 10)
+
+	gate := make(chan struct{})
+	entered := make(chan struct{}, 1)
+	held := make(chan error, 1)
+
+	openGate := sync.OnceFunc(func() { close(gate) })
+	t.Cleanup(openGate)
+
+	processInBackground(proc, &gatedReader{
+		data: bytes.NewReader(input), gate: gate, entered: entered,
+		counter: &readingCounter{},
+	}, held)
+	waitForEntries(t, entered, 1)
+
+	start := time.Now()
+
+	_, err := proc.Process(context.Background(), bytes.NewReader(input),
+		smallJPEGRequest())
+	if !errors.Is(err, ErrTooManyImages) {
+		t.Fatalf("Process() error = %v, want ErrTooManyImages", err)
+	}
+
+	if waited := time.Since(start); waited < proc.processingWaitTimeout {
+		t.Errorf("Process() failed after %v, before waiting %v",
+			waited, proc.processingWaitTimeout)
+	}
+
+	openGate()
+
+	err = <-held
+	if err != nil {
+		t.Errorf("Process() holding the slot: error = %v, want nil", err)
+	}
+}
+
+// TestProcessReleasesSlotOnError checks that Process gives its slot back
+// when it fails, whether it fails early or late: with one slot, the slot
+// must be free after the failure and the next call must succeed.
+func TestProcessReleasesSlotOnError(t *testing.T) {
+	t.Parallel()
+
+	valid := createTestJPEG(t, 10, 10)
+
+	unsupported := smallJPEGRequest()
+	unsupported.Format = "bmp"
+
+	cases := []struct {
+		name  string
+		input io.Reader
+		req   *Request
+		// want is the error Process must return; nil means any error.
+		want error
+	}{
+		{
+			name:  "input cannot be read",
+			input: iotest.ErrReader(errTestReadFailed),
+			req:   smallJPEGRequest(),
+			want:  errTestReadFailed,
+		},
+		{
+			name:  "input over the byte limit",
+			input: bytes.NewReader(createTestJPEG(t, 800, 600)),
+			req:   smallJPEGRequest(),
+			want:  ErrInputDataTooLarge,
+		},
+		{
+			name:  "input not an image",
+			input: strings.NewReader("not an image"),
+			req:   smallJPEGRequest(),
+		},
+		{
+			name:  "output format not supported",
+			input: bytes.NewReader(valid),
+			req:   unsupported,
+			want:  ErrUnsupportedOutputFormat,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			proc := New(Params{MaxInputBytes: 4096, MaxConcurrentProcessing: 1})
+			proc.processingWaitTimeout = 100 * time.Millisecond
+
+			_, err := proc.Process(context.Background(), tc.input, tc.req)
+			if err == nil || (tc.want != nil && !errors.Is(err, tc.want)) {
+				t.Fatalf("Process() error = %v, want %v", err, tc.want)
+			}
+
+			if held := len(proc.processingSemaphore); held != 0 {
+				t.Fatalf("slot still held after the error: %d held", held)
+			}
+
+			result, err := proc.Process(context.Background(), bytes.NewReader(valid),
+				smallJPEGRequest())
+			if err != nil {
+				t.Fatalf("Process() after the error = %v, want nil", err)
+			}
+
+			_ = result.Content.Close()
+		})
+	}
+}