pixa/internal/imgcache/concurrent_misses_internal_test.go

package imgcache

import (
	"bytes"
	"context"
	"errors"
	"image/jpeg"
	"io"
	"io/fs"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/getsentry/sentry-go"
	"sneak.berlin/go/pixa/internal/httpfetcher"
	"sneak.berlin/go/pixa/internal/imageprocessor"
	"sneak.berlin/go/pixa/internal/magic"
)

// arrivalWait is how long a test gives requests it has started to reach the
// point where they wait for a held fetch.
const arrivalWait = 100 * time.Millisecond

// heldFetcher counts the fetches it is asked for and holds each one until
// releaseFetches is called, so that a test can have requests arrive while a
// fetch is in progress. started receives once for every fetch.
type heldFetcher struct {
	upstream    httpfetcher.Fetcher
	fetches     atomic.Int32
	started     chan struct{}
	release     chan struct{}
	releaseOnce sync.Once
}

func (f *heldFetcher) Fetch(
	ctx context.Context, url string,
) (*httpfetcher.FetchResult, error) {
	f.fetches.Add(1)

	f.started <- struct{}{}

	select {
	case <-f.release:
	case <-ctx.Done():
		return nil, ctx.Err()
	}

	return f.upstream.Fetch(ctx, url)
}

// releaseFetches lets every held fetch, and every later one, go on.
func (f *heldFetcher) releaseFetches() {
	f.releaseOnce.Do(func() { close(f.release) })
}

// setupHeldFetchService returns a test service whose fetches go through a
// heldFetcher. Its database is limited to one connection: each connection to
// an in-memory SQLite database opens a new, empty one, so requests running at
// once must share the connection that holds the schema.
func setupHeldFetchService(t *testing.T) (*Service, *TestFixtures, *heldFetcher) {
	t.Helper()

	svc, fixtures := SetupTestService(t)
	svc.cache.db.SetMaxOpenConns(1)

	fetcher := &heldFetcher{
		upstream: svc.fetcher,
		started:  make(chan struct{}, 100),
		release:  make(chan struct{}),
	}
	svc.fetcher = fetcher

	t.Cleanup(fetcher.releaseFetches)

	return svc, fixtures, fetcher
}

// photoVariant asks for the test photo, 100x100, at 50x25 as a JPEG of the
// given quality and fit mode. Each call returns a new request, as Get writes
// to the request it is given.
func photoVariant(fixtures *TestFixtures, quality int, fit FitMode) *ImageRequest {
	return &ImageRequest{
		SourceHost: fixtures.GoodHost,
		SourcePath: testPathPhoto,
		Size:       Size{Width: 50, Height: 25},
		Format:     FormatJPEG,
		Quality:    quality,
		FitMode:    fit,
	}
}

// getResult is what one Get call returned, with the image read out.
type getResult struct {
	image []byte
	err   error
}

// startGet calls Get in a goroutine of its own and delivers what it returned
// on the channel.
func startGet(
	ctx context.Context, svc *Service, req *ImageRequest,
) <-chan getResult {
	results := make(chan getResult, 1)

	go func() {
		resp, err := svc.Get(ctx, req)
		if err != nil {
			results <- getResult{err: err}

			return
		}

		defer func() { _ = resp.Content.Close() }()

		image, err := io.ReadAll(resp.Content)
		results <- getResult{image: image, err: err}
	}()

	return results
}

// jpegSize returns the width and height of the JPEG image in data, or 0 and 0
// if data is not one.
func jpegSize(data []byte) (int, int) {
	config, err := jpeg.DecodeConfig(bytes.NewReader(data))
	if err != nil {
		return 0, 0
	}

	return config.Width, config.Height
}

// TestService_Get_ConcurrentMissesShareOneFetch starts several requests for
// one uncached variant while the first one's fetch is held. Between them they
// must fetch the source once and transcode it once, every one must be answered
// with the same 50x25 JPEG, and each must count one miss.
func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) {
	t.Parallel()

	svc, fixtures, fetcher := setupHeldFetchService(t)

	const requests = 8

	pending := make([]<-chan getResult, 0, requests)
	for range requests {
		pending = append(pending,
			startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover)))
	}

	<-fetcher.started
	time.Sleep(arrivalWait)
	fetcher.releaseFetches()

	var first []byte

	for i, results := range pending {
		got := <-results
		if got.err != nil {
			t.Fatalf("request %d: Get() error = %v", i, got.err)
		}

		if width, height := jpegSize(got.image); width != 50 || height != 25 {
			t.Errorf("request %d: image is %dx%d, want a 50x25 JPEG", i, width, height)
		}

		if first == nil {
			first = got.image
		} else if !bytes.Equal(got.image, first) {
			t.Errorf("request %d: image differs from request 0's", i)
		}
	}

	if fetches := fetcher.fetches.Load(); fetches != 1 {
		t.Errorf("%d requests made %d upstream fetches, want 1", requests, fetches)
	}

	// NewTestFS builds the same files the test service's fetcher serves.
	testFS, _ := NewTestFS(t)

	photo, err := fs.ReadFile(testFS, fixtures.GoodHostJPEG)
	if err != nil {
		t.Fatal(err)
	}

	want := cacheStatsCounters{0, requests, 1, int64(len(photo)), 1}

	if got := readCacheStatsCounters(t, svc.cache); got != want {
		t.Errorf("counters = %+v, want %+v", got, want)
	}
}

// TestService_Get_ConcurrentVariantsStayApart requests three variants of the
// test photo at once that differ only in quality or fit. Each must be made by
// a fetch and a transcode of its own, and each answer must be its own variant.
func TestService_Get_ConcurrentVariantsStayApart(t *testing.T) {
	t.Parallel()

	svc, fixtures, fetcher := setupHeldFetchService(t)

	cover := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))
	lowQuality := startGet(t.Context(), svc, photoVariant(fixtures, 40, FitCover))
	contain := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitContain))

	for range 3 {
		select {
		case <-fetcher.started:
		case <-time.After(5 * time.Second):
			t.Fatal("fewer fetches started than variants requested: " +
				"variants differing in quality or fit were merged")
		}
	}

	fetcher.releaseFetches()

	images := make(map[string][]byte)

	for _, variant := range []struct {
		name          string
		results       <-chan getResult
		width, height int
	}{
		{"q=85 fit=cover", cover, 50, 25},
		{"q=40 fit=cover", lowQuality, 50, 25},
		{"q=85 fit=contain", contain, 25, 25},
	} {
		got := <-variant.results
		if got.err != nil {
			t.Fatalf("%s: Get() error = %v", variant.name, got.err)
		}

		width, height := jpegSize(got.image)
		if width != variant.width || height != variant.height {
			t.Errorf("%s: image is %dx%d, want a %dx%d JPEG", variant.name,
				width, height, variant.width, variant.height)
		}

		images[variant.name] = got.image
	}

	if bytes.Equal(images["q=85 fit=cover"], images["q=40 fit=cover"]) {
		t.Error("q=40 was answered with the q=85 image")
	}
}

// TestService_Get_WaiterStopsWhenItsContextEnds has a second request for a
// variant join the first one's held fetch, then ends the second request's
// context. The second request must return at once with the context's error,
// while the fetch is still held, and the first must still be answered.
func TestService_Get_WaiterStopsWhenItsContextEnds(t *testing.T) {
	t.Parallel()

	svc, fixtures, fetcher := setupHeldFetchService(t)

	first := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))

	<-fetcher.started

	waiterCtx, cancelWaiter := context.WithCancel(t.Context())
	waiter := startGet(waiterCtx, svc, photoVariant(fixtures, 85, FitCover))

	time.Sleep(arrivalWait)
	cancelWaiter()

	select {
	case got := <-waiter:
		if !errors.Is(got.err, context.Canceled) {
			t.Errorf("waiting request: Get() error = %v, want %v",
				got.err, context.Canceled)
		}
	case <-time.After(time.Second):
		t.Fatal("waiting request did not return when its context ended")
	}

	fetcher.releaseFetches()

	got := <-first
	if got.err != nil {
		t.Fatalf("first request: Get() error = %v", got.err)
	}

	if width, height := jpegSize(got.image); width != 50 || height != 25 {
		t.Errorf("first request: image is %dx%d, want a 50x25 JPEG", width, height)
	}

	if fetches := fetcher.fetches.Load(); fetches != 1 {
		t.Errorf("upstream fetches = %d, want 1", fetches)
	}
}

// TestService_Get_FirstRequestLeavingKeepsTheWork ends the context of the
// request whose fetch is held, after a second request has joined it. The fetch
// and transcode must go on and answer the second request. The first request
// waits for its own work, as every request did before misses were shared, and
// is answered too.
func TestService_Get_FirstRequestLeavingKeepsTheWork(t *testing.T) {
	t.Parallel()

	svc, fixtures, fetcher := setupHeldFetchService(t)

	firstCtx, cancelFirst := context.WithCancel(t.Context())
	first := startGet(firstCtx, svc, photoVariant(fixtures, 85, FitCover))

	<-fetcher.started

	second := startGet(t.Context(), svc, photoVariant(fixtures, 85, FitCover))

	time.Sleep(arrivalWait)
	cancelFirst()
	fetcher.releaseFetches()

	for name, results := range map[string]<-chan getResult{
		"first request": first, "second request": second,
	} {
		got := <-results
		if got.err != nil {
			t.Fatalf("%s: Get() error = %v", name, got.err)
		}

		if width, height := jpegSize(got.image); width != 50 || height != 25 {
			t.Errorf("%s: image is %dx%d, want a 50x25 JPEG", name, width, height)
		}
	}

	if fetches := fetcher.fetches.Load(); fetches != 1 {
		t.Errorf("upstream fetches = %d, want 1", fetches)
	}
}

// TestService_Get_ConcurrentMissesShareAFailure has several requests for an
// image that cannot be served arrive while its fetch is held: the one fetch
// answers all of them with its error. The request after them is answered from
// the negative cache when the failure is kept there, and fetches again when it
// is not.
func TestService_Get_ConcurrentMissesShareAFailure(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name        string
		path        string
		wantErr     error // what every request at once gets
		wantNextErr error // what the request after them gets
		wantFetches int32 // fetches once the request after them is answered
	}{
		{"upstream answers 404, kept in the negative cache",
			"/images/missing.jpg", httpfetcher.ErrUpstreamError,
			ErrNegativeCached, 1},
		{"source fails the magic byte check, not kept",
			"/images/text.png", magic.ErrUnknownFormat, magic.ErrUnknownFormat, 2},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()

			svc, fixtures, fetcher := setupHeldFetchService(t)

			request := func() *ImageRequest {
				req := photoVariant(fixtures, 85, FitCover)
				req.SourcePath = tc.path

				return req
			}

			pending := make([]<-chan getResult, 0, 4)
			for range 4 {
				pending = append(pending, startGet(t.Context(), svc, request()))
			}

			<-fetcher.started
			time.Sleep(arrivalWait)
			fetcher.releaseFetches()

			for i, results := range pending {
				if got := <-results; !errors.Is(got.err, tc.wantErr) {
					t.Errorf("request %d: Get() error = %v, want %v", i, got.err, tc.wantErr)
				}
			}

			_, err := svc.Get(t.Context(), request())
			if !errors.Is(err, tc.wantNextErr) {
				t.Errorf("next request: Get() error = %v, want %v", err, tc.wantNextErr)
			}

			if fetches := fetcher.fetches.Load(); fetches != tc.wantFetches {
				t.Errorf("upstream fetches = %d, want %d", fetches, tc.wantFetches)
			}
		})
	}
}

// TestService_Get_EndedRequestFetchesNothing checks that a request whose
// context has already ended when it misses the cache starts no fetch.
func TestService_Get_EndedRequestFetchesNothing(t *testing.T) {
	t.Parallel()

	svc, fixtures, fetcher := setupHeldFetchService(t)

	ctx, cancel := context.WithCancel(t.Context())
	cancel()

	_, err := svc.Get(ctx, photoVariant(fixtures, 85, FitCover))
	if !errors.Is(err, context.Canceled) {
		t.Errorf("Get() error = %v, want %v", err, context.Canceled)
	}

	if fetches := fetcher.fetches.Load(); fetches != 0 {
		t.Errorf("upstream fetches = %d, want 0", fetches)
	}
}

// TestService_Get_ReturnsByItsDeadline gives a request a deadline and holds
// its fetch until the fetch's context ends, as the fetcher does while it waits
// for a free connection to the host. The request must return by its deadline
// with the deadline's error.
func TestService_Get_ReturnsByItsDeadline(t *testing.T) {
	t.Parallel()

	svc, fixtures, _ := setupHeldFetchService(t)

	const timeout = 200 * time.Millisecond

	ctx, cancel := context.WithTimeout(t.Context(), timeout)
	defer cancel()

	results := startGet(ctx, svc, photoVariant(fixtures, 85, FitCover))

	select {
	case got := <-results:
		if !errors.Is(got.err, context.DeadlineExceeded) {
			t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
		}
	case <-time.After(timeout + time.Second):
		t.Fatal("request did not return by its deadline")
	}
}

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

func (panickingFetcher) Fetch(
	context.Context, string,
) (*httpfetcher.FetchResult, error) {
	panic("upstream fetcher panicked")
}

// TestService_Get_PanicBecomesAnError checks that a panic while a variant is
// being made reaches its request as an error naming the panic, instead of
// being raised again.
func TestService_Get_PanicBecomesAnError(t *testing.T) {
	t.Parallel()

	svc, fixtures := SetupTestService(t)
	svc.fetcher = panickingFetcher{}

	var err error

	func() {
		defer func() {
			if recovered := recover(); recovered != nil {
				t.Fatalf("Get() panicked: %v", recovered)
			}
		}()

		_, err = svc.Get(t.Context(), photoVariant(fixtures, 85, FitCover))
	}()

	if err == nil || !strings.Contains(err.Error(), "upstream fetcher panicked") {
		t.Errorf("Get() error = %v, want one naming the panic", err)
	}
}

// TestService_Get_PanicIsReportedToSentry checks that a panic while a variant
// is being made is reported through the Sentry hub on the request's context,
// where the Sentry middleware puts one when sentry_dsn is set.
func TestService_Get_PanicIsReportedToSentry(t *testing.T) {
	t.Parallel()

	svc, fixtures := SetupTestService(t)
	svc.fetcher = panickingFetcher{}

	transport := &sentry.MockTransport{}

	client, err := sentry.NewClient(sentry.ClientOptions{
		Dsn:       "https://abc123@sentry.example.com/42",
		Transport: transport,
	})
	if err != nil {
		t.Fatal(err)
	}

	ctx := sentry.SetHubOnContext(t.Context(),
		sentry.NewHub(client, sentry.NewScope()))

	func() {
		defer func() {
			if recovered := recover(); recovered != nil {
				t.Fatalf("Get() panicked: %v", recovered)
			}
		}()

		_, _ = svc.Get(ctx, photoVariant(fixtures, 85, FitCover))
	}()

	events := transport.Events()
	if len(events) != 1 || events[0].Message != "upstream fetcher panicked" {
		t.Errorf("Sentry events = %+v, want one for the panic", events)
	}
}