pixa/internal/imageprocessor/max_concurrent_processing_internal_test.go

package imageprocessor

import (
	"bytes"
	"context"
	"errors"
	"io"
	"strings"
	"sync"
	"testing"
	"testing/iotest"
	"time"

	"sneak.berlin/go/pixa/internal/imagerequest"
)

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

// smallJPEGTransform asks for a 5x5 JPEG.
func smallJPEGTransform() *imagerequest.Transform {
	return &imagerequest.Transform{
		Size:    imagerequest.Size{Width: 5, Height: 5},
		Format:  imagerequest.FormatJPEG,
		Quality: 85,
		FitMode: imagerequest.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, smallJPEGTransform())
		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 TestNewMakesMaxConcurrentProcessingSlots(t *testing.T) {
	t.Parallel()

	proc := New(Params{MaxInputBytes: 50 << 20, 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{MaxInputBytes: 50 << 20, 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{MaxInputBytes: 50 << 20, 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),
		smallJPEGTransform())
	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 := smallJPEGTransform()
	unsupported.Format = "bmp"

	cases := []struct {
		name      string
		input     io.Reader
		transform *imagerequest.Transform
		// want is the error Process must return; nil means any error.
		want error
	}{
		{
			name:      "input cannot be read",
			input:     iotest.ErrReader(errTestReadFailed),
			transform: smallJPEGTransform(),
			want:      errTestReadFailed,
		},
		{
			name:      "input over the byte limit",
			input:     bytes.NewReader(createTestJPEG(t, 800, 600)),
			transform: smallJPEGTransform(),
			want:      ErrInputDataTooLarge,
		},
		{
			name:      "input not an image",
			input:     strings.NewReader("not an image"),
			transform: smallJPEGTransform(),
		},
		{
			name:      "output format not supported",
			input:     bytes.NewReader(valid),
			transform: 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.transform)
			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),
				smallJPEGTransform())
			if err != nil {
				t.Fatalf("Process() after the error = %v, want nil", err)
			}

			_ = result.Content.Close()
		})
	}
}

// TestWaitForProcessing holds a processing slot with a Process call that
// cannot finish reading its input. WaitForProcessing must report that image
// when its context ends first, wait for it otherwise, return 0 once it has
// finished, and give back the slots it took while waiting.
func TestWaitForProcessing(t *testing.T) {
	t.Parallel()

	proc := New(Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 2})

	gate := make(chan struct{})
	entered := make(chan struct{}, 1)
	results := make(chan error, 1)

	openGate := sync.OnceFunc(func() { close(gate) })
	t.Cleanup(openGate)

	processInBackground(proc, &gatedReader{
		data: bytes.NewReader(createTestJPEG(t, 10, 10)), gate: gate,
		entered: entered, counter: &readingCounter{},
	}, results)
	waitForEntries(t, entered, 1)

	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
	defer cancel()

	stillProcessing := proc.WaitForProcessing(ctx)
	t.Logf("WaitForProcessing() after its context ended: %d", stillProcessing)

	if stillProcessing != 1 {
		t.Errorf("WaitForProcessing() after its context ended = %d, want 1",
			stillProcessing)
	}

	waited := make(chan int, 1)

	go func() { waited <- proc.WaitForProcessing(t.Context()) }()

	select {
	case got := <-waited:
		t.Fatalf("WaitForProcessing() = %d while an image was being processed",
			got)
	case <-time.After(100 * time.Millisecond):
	}

	openGate()

	err := <-results
	if err != nil {
		t.Errorf("Process() error = %v, want nil", err)
	}

	select {
	case got := <-waited:
		if got != 0 {
			t.Errorf("WaitForProcessing() once processing finished = %d, want 0",
				got)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("WaitForProcessing() did not return once processing finished")
	}

	if held := len(proc.processingSemaphore); held != 0 {
		t.Errorf("%d slots still held after WaitForProcessing() returned", held)
	}
}