package ci
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
// deferralEngine is a dagnats that lists one worker with the given
// capacity, registers anything, finds nothing to supersede, and records
// which check-ins were started.
func deferralEngine(t *testing.T, maxTasks int, busy *int, started *[]string) *httptest.Server {
t.Helper()
return deferralEngineRuns(t, maxTasks, busy, 0, started)
}
// deferralEngineRuns is deferralEngine where every run it starts holds
// stepsPerRun running steps, as a real run does, and where the engine
// also lists a stale second registration of the worker, as it does for
// a while after the worker restarts.
func deferralEngineRuns(t *testing.T, maxTasks int, busy *int, stepsPerRun int, started *[]string) *httptest.Server {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/v1/workers":
fresh := time.Now().UTC().Format(time.RFC3339Nano)
fmt.Fprintf(w, `[{"worker_id":"w1","max_tasks":%d,"last_seen":%q},{"worker_id":"w0-before-restart","max_tasks":%d,"last_seen":"2020-01-01T00:00:00Z"}]`, maxTasks, fresh, maxTasks)
case r.URL.Path == "/workflows" || strings.HasPrefix(r.URL.Path, "/workflows/"):
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"status":"registered"}`))
case r.URL.Path == "/runs" && r.Method == http.MethodGet && r.URL.Query().Get("status") == "running":
// busy steps running, plus a run queued for days on a task
// type no worker serves: it must not count.
fmt.Fprint(w, `[`)
for i := 0; i < *busy+stepsPerRun*len(*started); i++ {
fmt.Fprintf(w, `{"run_id":"live-%d","status":"running","created_at":"2020-01-01T00:00:00Z","labels":{"repo":"alpha","pipeline":"ci/build.json"},"steps":{"test":{"status":"running"}}},`, i)
}
// And a workflow that is not CI at all: the notification
// service's digest, running for hours on the same engine.
fmt.Fprint(w, `{"run_id":"digest","status":"running","created_at":"2020-01-01T00:00:00Z","steps":{"S1":{"status":"running"}}},`)
fmt.Fprint(w, `{"run_id":"zombie","status":"running","created_at":"2020-01-01T00:00:00Z","labels":{"repo":"sandbox","pipeline":"ci/html.yml"},"steps":{"html":{"status":"queued"},"puzzles":{"status":"queued"}}}]`)
case r.URL.Path == "/runs" && r.Method == http.MethodGet:
fmt.Fprint(w, `[]`)
case r.URL.Path == "/runs":
var body struct {
Input map[string]any `json:"input"`
}
json.NewDecoder(r.Body).Decode(&body)
*started = append(*started, fmt.Sprint(body.Input["checkin"]))
w.WriteHeader(http.StatusCreated)
fmt.Fprintf(w, `{"run_id":"r-%d"}`, len(*started))
default:
http.NotFound(w, r)
}
}))
t.Cleanup(srv.Close)
return srv
}
func pipelined(news []Checkin, landings map[string]bool) *fakeRepo {
return &fakeRepo{
files: []string{"ci/build.json"},
blobs: map[string][]byte{"ci/build.json": []byte(`{"name":"build","version":"1.0","steps":[]}`)},
news: news,
landings: landings,
}
}
// WHEN THE WORKER IS FULL, BRANCH CHURN WAITS; TRUNK AND PROPOSED
// BRANCHES DO NOT; AND THE WAIT ENDS WHEN A SLOT FREES OR A LANDING
// OPENS.
//
// dmestas's rule for 029c4730, after the worker's own semaphore: a
// burst of pushes on side branches must not put a landing's run behind
// them in the worker's queue. So a check-in on a branch nobody has
// proposed is recorded as waiting instead of started while the workers
// are at capacity; trunk (what a deploy builds from) and any branch
// with an open landing start regardless; and the next round drains
// the waiting ones, longest wait first, as room appears — or at once
// for a branch that has since been proposed.
func TestAFullWorkerDefersBranchChurnAndDrainsIt(t *testing.T) {
var started []string
busy := 2
srv := deferralEngine(t, 2, &busy, &started)
st := &fakeStore{mark: map[string]string{"alpha": "old"}} // full: 2 of 2 steps running
repo := pipelined([]Checkin{
{Hash: "t1", Branch: "trunk"},
{Hash: "b1", Branch: "feat/quiet"},
{Hash: "b2", Branch: "feat/proposed"},
}, map[string]bool{"feat/proposed": true})
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := strings.Join(started, " "); got != "t1 b2" {
t.Fatalf("FAIL: started %q, want trunk and the proposed branch only", got)
}
if w, _ := st.Deferred("alpha"); len(w) != 1 || w[0].Hash != "b1" {
t.Fatalf("FAIL: waiting = %+v, want the quiet branch's check-in", w)
}
if m, _ := st.CIWatermark("alpha"); m != "b2" {
t.Errorf("FAIL: watermark %q: a deferred check-in must still advance it, or it would be re-read forever", m)
}
// Still full, nothing new: the wait holds.
repo.news = nil
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if len(started) != 2 {
t.Fatalf("FAIL: a full worker took the waiting check-in: %v", started)
}
// A slot frees: the wait ends, and the row goes with it.
busy = 1
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := strings.Join(started, " "); got != "t1 b2 b1" {
t.Fatalf("FAIL: started %q, want the waiting check-in launched once room appeared", got)
}
if w, _ := st.Deferred("alpha"); len(w) != 0 {
t.Errorf("FAIL: launched but still recorded as waiting: %+v", w)
}
t.Log("ok: churn waits, landings and trunk do not, and the wait ends with the room")
}
// A branch proposed while its check-in waited is launched at the next
// round even with no room: the landing is what the wait protects.
func TestAWaitingBranchStartsWhenItIsProposed(t *testing.T) {
var started []string
busy := 1
srv := deferralEngine(t, 1, &busy, &started)
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := pipelined([]Checkin{{Hash: "b1", Branch: "feat/x"}}, map[string]bool{})
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if len(started) != 0 {
t.Fatalf("FAIL: started %v with no room", started)
}
repo.news = nil
repo.landings["feat/x"] = true
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := strings.Join(started, " "); got != "b1" {
t.Fatalf("FAIL: started %q, want the proposed branch's check-in despite no room", got)
}
}
// A forge whose engine does not list workers never defers: unknown
// room is not full.
func TestUnknownCapacityNeverDefers(t *testing.T) {
var started []string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/v1/workers":
http.NotFound(w, r)
case r.URL.Path == "/workflows":
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"status":"registered"}`))
case r.URL.Path == "/runs" && r.Method == http.MethodGet:
fmt.Fprint(w, `[]`)
case r.URL.Path == "/runs":
var body struct {
Input map[string]any `json:"input"`
}
json.NewDecoder(r.Body).Decode(&body)
started = append(started, fmt.Sprint(body.Input["checkin"]))
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"run_id":"r-1"}`))
}
}))
defer srv.Close()
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := pipelined([]Checkin{{Hash: "b1", Branch: "feat/x"}}, map[string]bool{})
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if len(started) != 1 {
t.Fatalf("FAIL: with unknown capacity the check-in was deferred: started %v", started)
}
}
// A run queued for days on a task type no worker serves occupies no
// slot, and neither does a workflow that is not CI. Counting it is what made the first version of the deferral call
// a worker with one step running "full" (13 running runs against a
// capacity of 3 on 7a5d179697db, twelve of them dantest-sandbox runs
// waiting for html/puzzles workers that do not exist).
func TestAZombieRunDoesNotFillTheWorker(t *testing.T) {
var started []string
busy := 0
srv := deferralEngine(t, 3, &busy, &started)
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
repo := pipelined([]Checkin{{Hash: "b1", Branch: "feat/quiet"}}, map[string]bool{})
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if len(started) != 1 {
t.Fatalf("FAIL: a worker with nothing running deferred a branch: started %v", started)
}
if n, err := New(srv.URL).Occupied(time.Now()); err != nil || n != 0 {
t.Errorf("FAIL: occupied = %d, %v; the zombie's queued steps counted", n, err)
}
}
// A drain fills the worker by STEPS, and a stale registration adds no
// capacity. Measured on 7ead79c34e8f: four waiting branches all started
// in one second into a worker with three slots, because each launch
// reserved one slot while a quarry run holds three steps, and the
// engine still listed the worker's pre-restart registration.
func TestADrainFillsTheWorkerByStepsNotRuns(t *testing.T) {
var started []string
busy := 0
srv := deferralEngineRuns(t, 3, &busy, 3, &started)
st := &fakeStore{mark: map[string]string{"alpha": "old"}}
for _, h := range []string{"w1", "w2", "w3", "w4"} {
st.Defer("alpha", Checkin{Hash: h, Branch: "feat/" + h})
}
repo := pipelined(nil, map[string]bool{})
if err := Trigger(New(srv.URL), st, "alpha", repo, "http://x/r/alpha"); err != nil {
t.Fatal(err)
}
if got := strings.Join(started, " "); got != "w1" {
t.Fatalf("FAIL: started %q, want only the longest wait: one quarry-sized run fills three slots", got)
}
if w, _ := st.Deferred("alpha"); len(w) != 3 {
t.Errorf("FAIL: waiting = %+v, want three still waiting", w)
}
}