Files
dsh-web/internal/api/images_enqueue_concurrency_test.go
T

130 lines
3.8 KiB
Go

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)
}
}
}