mesh-catalog/modules/pacman/cmd/pacman-tools/keeper_hooks_test.go

package main

// The upgrade parts joined to the keeper's loop (keeperHooks): the log read into runs and old records pruned at
// the start, an ask for this node taken from `asks`, and the records pruned again a day later.

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

func TestTheKeeperRunsTheUpgradePartsAtItsStartOnItsAsksAndDaily(t *testing.T) {
	db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
	db.set("glibc 2.42-1\n", "")
	k, states, _, clock, _ := newTestKeeper(db)
	log := capturedLog(t)
	k.Machine.ReadFile = func(path string) ([]byte, error) {
		if path == PacmanLog {
			return []byte(log), nil
		}
		return nil, errNoFile
	}
	k.MakeHooks = keeperHooks
	k.Logf = t.Logf
	k.Config = KeeperConfig{RecordDays: 180}
	records := states.store(StateRecords)
	records.Put(NodeKey("node-a", "20250101T040000Z"), Record{Node: "node-a", Run: "20250101T040000Z"})
	records.Put(NodeKey("node-b", "20250101T040000Z"), Record{Node: "node-b", Run: "20250101T040000Z"})
	done := Run{Node: "node-a", Run: "20261003T040000Z", Kind: RunUpgrade, From: FromAsk, Ask: "ask-1",
		Steps: []Step{{Step: StepRefused, At: t0, Why: "a test"}}}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ended := make(chan error, 1)
	go func() { ended <- k.Run(ctx) }()
	until := func(what string, ok func() bool) {
		t.Helper()
		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) {
			if ok() {
				return
			}
		}
		t.Fatalf("never: %s", what)
	}
	until("the log read into runs", func() bool {
		keys, _ := states.store(StateRuns).Keys()
		return len(keys) > 1
	})
	until("the old record pruned at the start", func() bool {
		e, _ := records.Get(NodeKey("node-a", "20250101T040000Z"))
		return e == nil
	})
	if e, _ := records.Get(NodeKey("node-b", "20250101T040000Z")); e == nil {
		t.Error("another node's record was pruned")
	}

	// a run an ask started, ended; its ask left behind by a keeper that went away.
	states.store(StateRuns).Put(NodeKey("node-a", done.Run), done)
	asks := states.store(StateAsks)
	asks.Put(NodeKey("node-a", "ask-1"), Ask{ID: "ask-1", Node: "node-a", Act: ActUpgrade, At: t0})
	until("the ask a run names deleted", func() bool {
		e, _ := asks.Get(NodeKey("node-a", "ask-1"))
		return e == nil
	})

	records.Put(NodeKey("node-a", "20250102T040000Z"), Record{Node: "node-a", Run: "20250102T040000Z"})
	time.Sleep(20 * time.Millisecond)
	if e, _ := records.Get(NodeKey("node-a", "20250102T040000Z")); e == nil {
		t.Fatal("pruned before a day passed")
	}
	clock.Advance(UpdatesEvery)
	until("the old record pruned a day later", func() bool {
		e, _ := records.Get(NodeKey("node-a", "20250102T040000Z"))
		return e == nil
	})
	cancel()
	if err := <-ended; err == nil || !strings.Contains(err.Error(), "canceled") {
		t.Errorf("Run: %v", err)
	}
}

// roundKeeper is a keeper of node in a round of the made-up nodes order, its machine's time the keeper's clock.
func roundKeeper(t *testing.T, node string, order []string) (*Keeper, *fakeStates, *fakeClock) {
	t.Helper()
	db := &pacmanDB{sl: syncDBs, checkup: Ran{Status: 2}}
	db.set("glibc 2.42-1\n", "")
	k, states, _, clock, _ := newTestKeeper(db)
	k.Node = node
	k.Machine.Now = clock.Now
	k.MakeHooks = keeperHooks
	k.Logf = t.Logf
	k.Config = KeeperConfig{RecordDays: 180, RoundStartMinutes: 120, UpgradeOrder: order}
	k.AskRecord = func(string) (*stdio.StateEntry, error) { return nil, nil }
	return k, states, clock
}

func until(t *testing.T, what string, ok func() bool) {
	t.Helper()
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(time.Millisecond) {
		if ok() {
			return
		}
	}
	t.Fatalf("never: %s", what)
}

func roundNodeOf(s *fakeStore, round, node string) RoundNode {
	var rn RoundNode
	if e, _ := s.Get(RoundNodeKey(round, node)); e != nil {
		_ = json.Unmarshal(e.Value, &rn)
	}
	return rn
}

// 1841/AC6 through the keeper's loop: the first of the order opens the week's round at its start on Saturday
// 04:00 and starts its run in it; the round's parts get the operator channel's reader and the rounds state.
//
// ADR 0585 §2, §3: it ranks the mesh's mirrors inside the write that opens the round, before the round's key, which names
// the list's revision; and its keeper writes the list to its node's file.
func TestTheFirstNodesKeeperOpensTheRoundAndTakesItsTurn(t *testing.T) {
	k, states, _ := roundKeeper(t, "node-a", []string{"node-a", "node-b"})
	root := &fakeRoot{}
	k.Root = root
	servers := []string{"https://nl.mirror.example/archlinux/$repo/os/$arch"}
	k.Rank = func(conf []byte, round string) (MirrorList, error) {
		if string(conf) != "--sort rate\n" {
			t.Errorf("ranked with %q", conf)
		}
		return MirrorList{Servers: servers, RankedAt: t0, RankedBy: "node-a", Round: round}, nil
	}
	// The node's mirror list file holds what the root helper last wrote.
	k.Machine.ReadFile = func(path string) ([]byte, error) {
		switch w := root.written(); {
		case path == ReflectorConf:
			return []byte("--sort rate\n"), nil
		case path == Mirrorlist && len(w) > 0:
			return RenderMirrorlist(w[len(w)-1]), nil
		}
		return nil, errNoFile
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = k.Run(ctx) }()
	rounds := states.store(StateRounds)
	round := RoundID(t0)
	until(t, "the round opened", func() bool { e, _ := rounds.Get(round); return e != nil })
	until(t, "node-a's run named in the round", func() bool { return roundNodeOf(rounds, round, "node-a").Run != "" })
	if rn := roundNodeOf(rounds, round, "node-b"); rn.Run != "" {
		t.Errorf("node-b started in node-a's keeper: %+v", rn)
	}
	var opened Round
	e, _ := rounds.Get(round)
	_ = json.Unmarshal(e.Value, &opened)
	list, rev, err := readMirrorList(states.store(StateMirrors))
	if err != nil || list == nil || list.Round != round || opened.Mirrors != rev || rev == 0 || opened.MirrorsNote != "" {
		t.Errorf("the round opened with mirrors %d (%q), the list is %+v at %d (%v)", opened.Mirrors, opened.MirrorsNote, list, rev, err)
	}
	until(t, "the list written to the node's file", func() bool { w := root.written(); return len(w) == 1 && w[0][0] == servers[0] })
	// The opener's run starts only once its file holds the round's list (written before the run, not refused for it).
	var run Run
	if e, _ := states.store(StateRuns).Get(NodeKey("node-a", roundNodeOf(rounds, round, "node-a").Run)); e != nil {
		_ = json.Unmarshal(e.Value, &run)
	}
	if lastStep(run) == StepRefused && strings.Contains(run.Steps[len(run.Steps)-1].Why, "mirror") {
		t.Errorf("node-a's run was refused for its mirror list: %+v", run.Steps)
	}
	if len(root.written()) != 1 {
		t.Errorf("the list was written %d times", len(root.written()))
	}
}

// 1841/AC6 through the keeper's loop: a node waits until the node before it passed its check, and starts on
// the change of `runs` that says so, with no wait for a timer.
func TestTheNextNodesKeeperStartsWhenTheRunBeforeItPassed(t *testing.T) {
	k, states, clock := roundKeeper(t, "node-b", []string{"node-a", "node-b"})
	round := RoundID(t0)
	rounds, runs := states.store(StateRounds), states.store(StateRuns)
	runA := RunID(t0)
	rounds.Put(round, Round{Round: round, Opened: t0, OpenedBy: "node-a", Order: []string{"node-a", "node-b"}})
	rounds.Put(RoundNodeKey(round, "node-a"), RoundNode{Round: round, Node: "node-a", Run: runA})
	a := Run{Node: "node-a", Run: runA, Kind: RunUpgrade, From: FromRound, Round: round,
		Steps: []Step{{Step: StepRecorded, At: t0}, {Step: StepStarted, At: t0}}}
	runs.Put(NodeKey("node-a", runA), a)
	clock.Advance(10 * time.Minute)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() { _ = k.Run(ctx) }()
	time.Sleep(30 * time.Millisecond)
	if rn := roundNodeOf(rounds, round, "node-b"); rn.Run != "" {
		t.Fatalf("node-b started before node-a passed: %+v", rn)
	}
	a.Steps = append(a.Steps, Step{Step: StepFinished, At: clock.Now()}, Step{Step: StepCheckedAfter, At: clock.Now(), Check: &Check{Passed: true}})
	runs.Put(NodeKey("node-a", runA), a)
	until(t, "node-b's run named in the round", func() bool { return roundNodeOf(rounds, round, "node-b").Run != "" })
}

// The pass of a node is answered from the operator channel's record: the keeper hands its reader on.
func TestTheKeeperHandsTheOperatorChannelsReaderToTheRound(t *testing.T) {
	k, states, _ := roundKeeper(t, "node-a", nil)
	k.State = states.State
	h := keeperHooks(k)
	if k.upgrades == nil || k.upgrades.AskRecord == nil || k.upgrades.Rounds == nil {
		t.Fatalf("the round's parts lack the channel's reader or the rounds state: %+v", k.upgrades)
	}
	if len(h.Wakes) < 2 {
		t.Fatalf("no wake for the round: %d wakes", len(h.Wakes))
	}
	// ADR 0585 §3: the node's file is written from the shared list when they differ, and never from a list that is empty
	// or unreadable; the file is written again when it differs (the watch's MirrorsRewritten).
	if len(h.MirrorsRewritten) != 1 {
		t.Fatalf("the mirror list file is not written again when it changes: %d hooks", len(h.MirrorsRewritten))
	}
	root := &fakeRoot{}
	u := k.upgrades
	u.Root = root
	file := []byte(nil)
	u.Machine.ReadFile = func(path string) ([]byte, error) {
		if path == Mirrorlist && file != nil {
			return file, nil
		}
		return nil, errNoFile
	}
	mirrors := states.store(StateMirrors)
	good := []string{"https://a.example/$repo/os/$arch", "https://b.example/archlinux/$repo/os/$arch"}
	for _, step := range []struct {
		name   string
		put    any
		file   []byte
		wrote  bool
		failed bool
	}{
		{name: "no shared list yet writes nothing"},
		{name: "an empty list is never written", put: MirrorList{Servers: []string{}}, failed: true},
		{name: "an unreadable list is never written", put: "not a list", failed: true},
		{name: "a list the file differs from is written whole", put: MirrorList{Servers: good}, file: []byte(mirrorlistReflector), wrote: true},
		{name: "a file holding the list is not written again", put: MirrorList{Servers: good}, file: RenderMirrorlist(good)},
	} {
		if step.put != nil {
			mirrors.Put(MirrorsKey, step.put)
		}
		file = step.file
		before := len(root.written())
		wrote, err := u.WriteMirrors()
		if wrote != step.wrote || (err != nil) != step.failed || len(root.written()) != before+map[bool]int{true: 1}[step.wrote] {
			t.Errorf("%s: wrote %v (%v), the root helper wrote %v", step.name, wrote, err, root.written())
		}
	}
	// ADR 0585 §2: a round's run starts only once the node's file holds the list the round opened with: none asked of a
	// round that opened with no list; refused, naming why, when the bus holds another revision or the file is not written.
	rounds0 := states.store(StateRounds)
	_, listRev, _ := readMirrorList(mirrors)
	file = RenderMirrorlist(good)
	for _, c := range []struct {
		name, refusal string
		opened        uint64
		fail          error
	}{
		{name: "a round that opened with no list", opened: 0},
		{name: "a round whose list the file holds", opened: listRev},
		{name: "a round that opened with another revision", opened: listRev + 7, refusal: "the bus holds revision"},
		{name: "a file that cannot be written", opened: listRev, fail: errors.New("no root"), refusal: "is not written"},
	} {
		rounds0.Put("2026-W40", Round{Round: "2026-W40", Opened: t0, Order: []string{"node-a"}, Mirrors: c.opened})
		if c.fail != nil {
			file = []byte(mirrorlistReflector)
		}
		root.fail = c.fail
		if why := u.roundMirrorsWritten("2026-W40"); (c.refusal == "") != (why == "") || !strings.Contains(why, c.refusal) {
			t.Errorf("%s: %q", c.name, why)
		}
	}
	root.fail = nil
	rounds0.Delete("2026-W40")
	// ADR 0585 §4: a refresh asked of the keeper is refused while a round runs, and ranks the list when none does; the
	// ask is taken once either way. A round that opens while it ranks refuses its write, naming the round.
	asks, rounds := states.store(StateAsks), states.store(StateRounds)
	ranked := 0
	u.Rank = func([]byte, string) (MirrorList, error) { ranked++; return MirrorList{Servers: good}, nil }
	u.Machine.ReadFile = func(string) ([]byte, error) { return []byte("--sort rate\n"), nil }
	rounds.Put("2026-W43", Round{Round: "2026-W43", Opened: t0, OpenedBy: "node-a", Order: []string{"node-a"}})
	for _, running := range []bool{true, false} {
		if !running {
			rounds.Delete("2026-W43")
		}
		key := NodeKey("node-a", "ask-m")
		asks.Put(key, Ask{ID: "ask-m", Node: "node-a", Act: ActMirrors, At: t0})
		e, _ := asks.Get(key)
		err := u.TakeAsk(stdio.StateChange{Op: "put", Key: key, Value: e.Value})
		if left, _ := asks.Get(key); left != nil || (err != nil) != running || ranked != map[bool]int{true: 0, false: 1}[running] ||
			(running && !strings.Contains(err.Error(), "round 2026-W43 runs")) {
			t.Errorf("a refresh with a round running %v: %v, ranked %d, ask left %v", running, err, ranked, left)
		}
	}
	_, before, _ := readMirrorList(mirrors)
	u.Rank = func([]byte, string) (MirrorList, error) {
		rounds.Put("2026-W44", Round{Round: "2026-W44", Opened: t0, OpenedBy: "node-a", Order: []string{"node-a"}})
		return MirrorList{Servers: good}, nil
	}
	asks.Put(NodeKey("node-a", "ask-r"), Ask{ID: "ask-r", Node: "node-a", Act: ActMirrors, At: t0})
	e, _ := asks.Get(NodeKey("node-a", "ask-r"))
	err := u.TakeAsk(stdio.StateChange{Op: "put", Key: NodeKey("node-a", "ask-r"), Value: e.Value})
	if _, after, _ := readMirrorList(mirrors); err == nil || !strings.Contains(err.Error(), "round 2026-W44 runs") || after != before {
		t.Errorf("a refresh whose round opened while it ranked: %v, the list at %d (was %d)", err, after, before)
	}
}