package api import ( "net/http" "sync" "testing" "time" ) func TestEnqueueReleaseConcurrentMissingImageReturnsReplacement(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": "0.2.0-rc.1", "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", "0.2.0-rc.1") info := releaseFixture() if _, err := a.s.DB.Exec("UPDATE image_builds SET release_id=?,source_repository=?,source_commit=?,version=? WHERE image_id='old-image'", info.ReleaseID, info.Repository, info.Commit, info.Version); err != nil { t.Fatal(err) } 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() var submissions sync.WaitGroup t.Cleanup(func() { close(stop) 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.enqueueRelease("u", info) 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) } } }