pixa/internal/imgcache/max_concurrent_processing_internal_test.go

package imgcache

import (
	"image/color"
	"image/jpeg"
	"io"
	"os"
	"testing"
	"time"

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

// 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.Request {
	return &imagerequest.Request{
		Source: imagerequest.Source{
			Host: fixtures.GoodHost,
			Path: testPathPhoto,
		},
		Transform: imagerequest.Transform{
			Size:    imagerequest.Size{Width: width},
			Format:  imagerequest.FormatJPEG,
			Quality: 85,
			FitMode: imagerequest.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, &imagerequest.Transform{})
	}()

	_, 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)
	}
}