dsh-web/internal/api/images_enqueue_concurrency_test.go

package api

import (
	"context"
	"net/http"
	"sync"
	"testing"
	"time"
)

func TestEnqueueBuildConcurrentMissingImageReturnsReplacement(t *testing.T) {
	inspections := make(chan chan struct{}, 2)
	stop := make(chan struct{})
	a, _ := imageTestAPI(t, func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodGet {
			t.Errorf("unexpected Docker method: %s", r.Method)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		switch r.URL.Path {
		case "/v1.47/images/json":
			writeJSON(w, []map[string]any{{"Id": "old-image", "Labels": map[string]string{
				"dsh-mgr.role": "runtime", "dsh-mgr.version": "1.2.3", "dsh-mgr.recipe": imageRecipeRev,
			}}})
		case "/v1.47/images/old-image/json":
			release := make(chan struct{})
			select {
			case inspections <- release:
			case <-stop:
				return
			}
			select {
			case <-release:
				w.WriteHeader(http.StatusNotFound)
			case <-stop:
			case <-r.Context().Done():
			}
		default:
			t.Errorf("unexpected Docker path: %s", r.URL.Path)
			w.WriteHeader(http.StatusInternalServerError)
		}
	})
	seedVerifiedImage(t, a, "old-image", "1.2.3")
	var oldID string
	if err := a.s.DB.QueryRow("SELECT id FROM image_builds WHERE image_id='old-image'").Scan(&oldID); err != nil {
		t.Fatal(err)
	}
	// Keep the replacement queued without starting a background worker.
	buildQueue.Lock()
	buildQueue.active[a.s.DB] = true
	buildQueue.Unlock()
	ctx, cancel := context.WithCancel(context.Background())
	var submissions sync.WaitGroup
	t.Cleanup(func() {
		close(stop)
		cancel()
		submissions.Wait()
		buildQueue.Lock()
		delete(buildQueue.active, a.s.DB)
		delete(buildQueue.initialized, a.s.DB)
		buildQueue.Unlock()
	})
	type result struct {
		build buildRecord
		err   error
	}
	submit := func() <-chan result {
		done := make(chan result, 1)
		submissions.Add(1)
		go func() {
			defer submissions.Done()
			b, err := a.enqueueBuild(ctx, "u", "1.2.3")
			done <- result{b, err}
		}()
		return done
	}
	waitInspection := func() chan struct{} {
		t.Helper()
		select {
		case release := <-inspections:
			return release
		case <-time.After(3 * time.Second):
			t.Fatal("submission did not inspect the old image")
			return nil
		}
	}
	waitResult := func(done <-chan result) buildRecord {
		t.Helper()
		select {
		case got := <-done:
			if got.err != nil {
				t.Fatal(got.err)
			}
			return got.build
		case <-time.After(3 * time.Second):
			t.Fatal("submission did not finish")
			return buildRecord{}
		}
	}
	firstDone := submit()
	firstRelease := waitInspection()
	secondDone := submit()
	secondRelease := waitInspection()
	close(firstRelease)
	first := waitResult(firstDone)
	if first.ID == "" || first.ID == oldID || first.Status != "queued" {
		t.Fatalf("first submission did not queue a replacement: %+v", first)
	}
	close(secondRelease)
	second := waitResult(secondDone)
	if second.ID != first.ID || second.Status != "queued" {
		t.Errorf("second submission returned id=%q status=%q; want replacement id=%q status=queued", second.ID, second.Status, first.ID)
	}
	var status string
	if err := a.s.DB.QueryRow("SELECT status FROM image_builds WHERE id=?", oldID).Scan(&status); err != nil || status != "missing" {
		t.Errorf("old build status=%q, err=%v; want missing", status, err)
	}
	for query, want := range map[string]int{
		"SELECT count(*) FROM image_builds":                       2,
		"SELECT count(*) FROM image_builds WHERE status='queued'": 1,
		"SELECT count(*) FROM image_build_requests":               1,
	} {
		var count int
		if err := a.s.DB.QueryRow(query).Scan(&count); err != nil || count != want {
			t.Errorf("%s: count=%d, err=%v; want %d", query, count, err, want)
		}
	}
}