package main
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
)
// drainRepo is a repository with one pipeline and nothing new.
type drainRepo struct{}
func (drainRepo) NewCheckins(string, int) ([]ci.Checkin, error) { return nil, nil }
func (drainRepo) Files(string) ([]string, error) { return []string{"ci/build.json"}, nil }
func (drainRepo) Read(string, string) ([]byte, error) {
return []byte(`{"name":"build","version":"1.0","steps":[]}`), nil
}
func (drainRepo) HasOpenLanding(string) bool { return false }
func (drainRepo) DefaultBranch() string { return "trunk" }
func (drainRepo) BranchOf(string) string { return "" }
// drainEngine lists one worker of one slot, busy while *busy is set,
// and records the check-ins it is asked to start.
func drainEngine(t *testing.T, busy *bool) (*ci.Engine, *[]string) {
t.Helper()
var mu sync.Mutex
started := &[]string{}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
switch {
case r.URL.Path == "/v1/workers":
fmt.Fprintf(w, `[{"worker_id":"w1","max_tasks":1,"last_seen":%q}]`, time.Now().UTC().Format(time.RFC3339Nano))
case 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":
if *busy || len(*started) > 0 {
fmt.Fprint(w, `[{"run_id":"live","status":"running","created_at":"2020-01-01T00:00:00Z","labels":{"repo":"test","pipeline":"ci/build.json"},"steps":{"s":{"status":"running"}}}]`)
} else {
fmt.Fprint(w, `[]`)
}
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 ci.New(srv.URL), started
}
// A check-in waiting for a worker slot is released by the timer's
// drain when room appears, with no store change to prompt it, and only
// on the node holding the store's lease (quarry#296).
func TestDrainDeferredReleasesAQuietRepositoryOnTheHolderOnly(t *testing.T) {
n := newJobNodes(t, lease.DefaultTTL, "node-a", "node-b")
holder, other := 0, 1
if !n.gates[holder].store("test") || n.gates[other].store("test") {
t.Fatal("FAIL: the first node to ask does not alone hold the store")
}
if err := n.db.DeferCI(oltp.CIDeferred{Repo: "test", Checkin: "b1", Branch: "feat/quiet"}); err != nil {
t.Fatal(err)
}
busy := true
engine, started := drainEngine(t, &busy)
mk := func(i int) *registrar {
rg := n.registrar(t, i)
rg.engine = engine
rg.drainRepo = func(string) (ci.Repo, bool) { return drainRepo{}, true }
rg.deps = &api.Deps{Repos: api.NewRepos(nil)}
return rg
}
mk(holder).drainDeferred()
if len(*started) != 0 {
t.Fatalf("FAIL: a full worker took the waiting check-in: %v", *started)
}
busy = false
mk(other).drainDeferred()
if len(*started) != 0 {
t.Fatalf("FAIL: the non-holder drained: %v", *started)
}
if w, _ := n.db.CIDeferred("test"); len(w) != 1 {
t.Fatalf("FAIL: the non-holder changed the waiting rows: %+v", w)
}
mk(holder).drainDeferred()
if got := strings.Join(*started, " "); got != "b1" {
t.Fatalf("FAIL: started %q, want the waiting check-in released by the drain", got)
}
if w, _ := n.db.CIDeferred("test"); len(w) != 0 {
t.Errorf("FAIL: launched but still waiting: %+v", w)
}
}