internal/imgcache/concurrent_misses_internal_test.go

@@ -0,0 +1,444 @@
+package imgcache
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"image/jpeg"
+	"io"
+	"io/fs"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/pixa/internal/httpfetcher"
+	"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)
+	}
+}
+
+// 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)
+	}
+}