package main
import (
"context"
"path/filepath"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/zitcli"
)
// jobNodes is one store, one shared OLTP database and one coordinator,
// with a job gate per node over a lease manager of its own.
type jobNodes struct {
db *oltp.DB
dir string
ix *index.Index
gates []*jobGate
mgrs []*lease.Manager
}
func newJobNodes(t *testing.T, ttl time.Duration, nodes ...string) *jobNodes {
t.Helper()
dir, ix := acceptedLanding(t)
db, err := oltp.Open(filepath.Join(t.TempDir(), "o.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
ev, err := events.Start(filepath.Join(t.TempDir(), "coord"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
code := func(repo string) (string, bool) {
if repo == "test" {
return "code-test", true
}
return "", false
}
n := &jobNodes{db: db, dir: dir, ix: ix}
for _, node := range nodes {
m, err := lease.New(ev.JS(), node, ttl)
if err != nil {
t.Fatal(err)
}
t.Cleanup(m.Close)
n.mgrs = append(n.mgrs, m)
n.gates = append(n.gates, &jobGate{leases: m, code: code})
}
return n
}
// registrar is node i's view of the store, with CI enabled. The engine
// is never reached: a store's first CI round only sets the watermark.
func (n *jobNodes) registrar(t *testing.T, i int) *registrar {
t.Helper()
return ®istrar{odb: n.db, engine: ci.New("http://127.0.0.1:1"),
deps: &api.Deps{Repos: api.NewRepos(nil)}, jobs: n.gates[i]}
}
// mirrorDeps serves the store with a Mirror that counts its pulls.
func (n *jobNodes) mirrorDeps(pulls *int) *api.Deps {
return &api.Deps{
Repos: api.NewRepos(map[string]*api.Repo{"test": {
Zit: zitcli.New(n.dir),
Mirror: func(string) (bool, error) { *pulls++; return false, nil },
}}),
Mirrors: mirrorStore{db: n.db},
}
}
func (n *jobNodes) bridges(i int) *bridgeRunner {
return &bridgeRunner{odb: n.db, code: n.gates[i].code, holds: n.gates[i].store,
running: map[string]*runningBridge{}}
}
type armed []oltp.AutoAccept
func (a armed) AutoAccepts() ([]oltp.AutoAccept, error) { return a, nil }
func (armed) ClearAutoAccept(string, string) error { return nil }
func (armed) SetAutoWaiting(string, string, string) {}
// Two nodes over one store: only the lease holder triggers CI, pulls
// the mirror, bridges chat and sweeps auto-accepts; the other skips
// them and leaves no trace of having tried.
func TestOnlyTheStoreHolderRunsItsJobs(t *testing.T) {
n := newJobNodes(t, lease.DefaultTTL, "node-a", "node-b")
holder, other := 0, 1
if !n.gates[holder].store("test") {
t.Fatal("the first node to ask does not hold the store")
}
if n.gates[other].store("test") {
t.Fatal("FAIL: both nodes hold the store")
}
cli := zitcli.New(n.dir)
// CI trigger: the watermark is the trigger's first write.
n.registrar(t, other).afterFold("test", cli, n.ix)
if _, ok := n.db.CIWatermark("test"); ok {
t.Fatal("FAIL: the non-holder's CI trigger fired")
}
n.registrar(t, holder).afterFold("test", cli, n.ix)
if _, ok := n.db.CIWatermark("test"); !ok {
t.Fatal("FAIL: the holder's CI trigger did not fire")
}
// Mirror pull: the non-holder neither pulls nor records a run.
if err := n.db.SetMirror("test", "https://example.invalid/test", time.Hour); err != nil {
t.Fatal(err)
}
pulls := 0
mirrorRound(n.mirrorDeps(&pulls), n.db, n.gates[other], time.Now())
if pulls != 0 {
t.Fatalf("FAIL: the non-holder pulled the mirror %d time(s)", pulls)
}
row, ok, err := n.db.MirrorFor("test")
if err != nil || !ok {
t.Fatalf("mirror row: ok=%v err=%v", ok, err)
}
if row.LastAt != "" || row.LastErr != "" || row.Failures != 0 {
t.Fatalf("FAIL: the non-holder left a run on the mirror row: at=%q err=%q failures=%d",
row.LastAt, row.LastErr, row.Failures)
}
mirrorRound(n.mirrorDeps(&pulls), n.db, n.gates[holder], time.Now())
if pulls != 1 {
t.Fatalf("FAIL: the holder pulled %d time(s), want 1", pulls)
}
if row, _, _ = n.db.MirrorFor("test"); row.Failures != 0 || row.LastErr != "" {
t.Fatalf("FAIL: the holder's pull recorded a failure: %q", row.LastErr)
}
// Chat bridge: started only where the store is held.
if err := n.db.SetChatBridge("test", "https://example.org/repo", "bot", "s3cret"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
nb, hb := n.bridges(other), n.bridges(holder)
nb.reconcile(ctx)
hb.reconcile(ctx)
if _, on := nb.running["test"]; on {
t.Fatal("FAIL: the non-holder started the chat bridge")
}
if _, on := hb.running["test"]; !on {
t.Fatal("FAIL: the holder did not start the chat bridge")
}
// Auto-accept sweep: the non-holder sees no armed landings.
rows := armed{{Repo: "test", Landing: "L1", Account: "dan"}}
if got, _ := (heldSweeper{AutoSweeper: rows, jobs: n.gates[other]}).AutoAccepts(); len(got) != 0 {
t.Fatalf("FAIL: the non-holder would sweep %d armed landing(s)", len(got))
}
if got, _ := (heldSweeper{AutoSweeper: rows, jobs: n.gates[holder]}).AutoAccepts(); len(got) != 1 {
t.Fatalf("FAIL: the holder would sweep %d armed landing(s), want 1", len(got))
}
// A holder that closes stops its bridge at the next reconcile.
n.mgrs[holder].Close()
hb.reconcile(ctx)
if _, on := hb.running["test"]; on {
t.Fatal("FAIL: a node that no longer holds the store kept its bridge")
}
t.Log("ok: trigger, mirror, bridge and sweep ran on the holder only")
}
// A fleet job runs on exactly one node and moves to another when that
// node closes.
func TestAFleetJobMovesWhenTheHolderCloses(t *testing.T) {
ttl := time.Second
n := newJobNodes(t, ttl, "node-a", "node-b")
a, b := n.gates[0], n.gates[1]
if !a.fleet() {
t.Fatal("the first node to ask does not run fleet jobs")
}
if b.fleet() {
t.Fatal("FAIL: two nodes run fleet jobs")
}
n.mgrs[0].Close()
if a.fleet() {
t.Fatal("FAIL: a closed node still runs fleet jobs")
}
deadline := time.Now().Add(5 * ttl)
for !b.fleet() {
if time.Now().After(deadline) {
t.Fatal("FAIL: fleet jobs never moved to the remaining node")
}
time.Sleep(50 * time.Millisecond)
}
t.Log("ok: fleet jobs moved to the remaining node")
}
// One node holds every lease, so every job still runs.
func TestASingleNodeSkipsNoJob(t *testing.T) {
n := newJobNodes(t, lease.DefaultTTL, "node-a")
g := n.gates[0]
if !g.fleet() || !g.store("test") {
t.Fatalf("FAIL: a lone node skips jobs: fleet=%v store=%v", g.fleet(), g.store("test"))
}
// A repository this node has no code for is not skipped either: its
// job runs and reports what it finds, as it did before placement.
if !g.store("unregistered") {
t.Fatal("FAIL: a repository without a code was skipped")
}
n.registrar(t, 0).afterFold("test", zitcli.New(n.dir), n.ix)
if _, ok := n.db.CIWatermark("test"); !ok {
t.Fatal("FAIL: the CI trigger did not fire on a single node")
}
if err := n.db.SetMirror("test", "https://example.invalid/test", time.Hour); err != nil {
t.Fatal(err)
}
pulls := 0
mirrorRound(n.mirrorDeps(&pulls), n.db, g, time.Now())
if pulls != 1 {
t.Fatalf("FAIL: the mirror was pulled %d time(s) on a single node, want 1", pulls)
}
if err := n.db.SetChatBridge("test", "https://example.org/repo", "bot", "s3cret"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
br := n.bridges(0)
br.reconcile(ctx)
if _, on := br.running["test"]; !on {
t.Fatal("FAIL: the chat bridge did not start on a single node")
}
rows := armed{{Repo: "test", Landing: "L1", Account: "dan"}}
if got, _ := (heldSweeper{AutoSweeper: rows, jobs: g}).AutoAccepts(); len(got) != 1 {
t.Fatalf("FAIL: the auto-accept sweep saw %d armed landing(s), want 1", len(got))
}
// No coordinator at all (no repositories served): nothing is gated.
var none *jobGate
if !none.fleet() || !none.store("test") {
t.Fatal("FAIL: a node with no lease manager skips jobs")
}
t.Log("ok: every job ran on a single node")
}