diff --git a/internal/imgcache/concurrent_misses_internal_test.go b/internal/imgcache/concurrent_misses_internal_test.go new file mode 100644 index 0000000..53b2db5 --- /dev/null +++ b/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) + } +}