mesh-controller/cmd/mesh-controller/rollout_events_test.go

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
	"github.com/novox/mesh-controller/internal/testbus"
)

// Acceptance of the step (novox/hq ADR 0406 §5, §6, acceptance 7; issue 974), live, with the store and a real bus and no
// ticker: a rollout of two tiers advances on the events alone — the plan kept, the machines' reports, a silent key's
// expiry — a machine that reports in time is never named silent and one that never reports is named silent when its
// key expires, and a controller replaced in the middle of it resumes with no tier asked twice and none skipped.
func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	inv := open.inventory

	// A quarter of an hour of silence is two seconds here.
	wasSilent := applySilentAfter
	applySilentAfter = 2 * time.Second
	t.Cleanup(func() { applySilentAfter = wasSilent })

	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := broker.AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	legacy, err := js.Conn().JetStream()
	if err != nil {
		t.Fatal(err)
	}
	checkEvents = link.OverNATS{Conn: js.Conn(), JS: legacy}
	t.Cleanup(func() { checkEvents = nil })
	sentHeard := make(chan *nats.Msg, 32)
	sub, err := js.Conn().ChanSubscribe(link.SeatEventSubject(link.MeshControllerSeat, link.KeySent), sentHeard)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })

	// Every save of a plan is kept, said and looked at by the serving controller's own hook.
	saved := &movesSaid{revs: map[string][]int64{}}
	wasSaved := inventory.PlanSaved
	serving := servingPlanSaved(ctx, checkEvents, open)
	inventory.PlanSaved = func(i *inventory.Inventory, p inventory.Plan) { saved.hear(i, p); serving(i, p) }
	t.Cleanup(func() { inventory.PlanSaved = wasSaved })

	// The build machine and the machines, as far as the plan meets them: an ask is counted, a send is recorded and kept.
	var mu sync.Mutex
	var sends [][]string
	asked := asksRecorded(t) // each module taken from a done key as the walk reaches it, said as repository#build id
	digests := 0
	wasSend := sendRollout
	sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
		mu.Lock()
		defer mu.Unlock()
		sends = append(sends, append([]string(nil), names...))
		current, err := open.inventory.CurrentBuilds(ctx)
		if err != nil {
			return nil, err
		}
		builds := map[string]string{}
		for _, m := range []string{"a", "b"} {
			builds[m] = current[m].Commit
		}
		for _, node := range names {
			digests++
			if err := open.inventory.RecordSent(ctx, nodeID(t, open, node), fmt.Sprintf("d-%s-%d", node, digests), builds); err != nil {
				return nil, err
			}
		}
		return names, nil
	}
	t.Cleanup(func() { sendRollout = wasSend })

	register := func(m, commit string, asked time.Time) {
		t.Helper()
		if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: commit},
			inventory.Source{Repository: "novox/" + m, Seat: "git", Path: "modules/" + m, BuiltFrom: commit, Head: commit,
				Asked: asked}); err != nil {
			t.Fatal(err)
		}
	}
	record := func(id, m, commit string, at time.Time) {
		t.Helper()
		b := inventory.Build{ID: id, Module: m, Commit: commit, Repository: "novox/" + m, Path: "modules/" + m, Asked: at, At: at}
		b.Manifest, _ = json.Marshal(catalogue.Manifest{Module: m, Version: commit})
		if err := inv.RecordBuild(ctx, b); err != nil {
			t.Fatal(err)
		}
	}
	old := time.Now().Add(-2 * time.Hour)
	for _, m := range []string{"a", "b"} {
		record("build-"+m+"-1", m, "c1", old)
		register(m, "c1", old)
		for _, n := range []string{"anchor", "laptop"} {
			if _, err := inv.Assign(ctx, n, m); err != nil {
				t.Fatal(err)
			}
		}
	}
	for _, n := range []string{"anchor", "laptop"} {
		if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{"a": "c1", "b": "c1"}); err != nil {
			t.Fatal(err)
		}
	}
	record("build-a-2", "a", "c2", time.Now().Add(-time.Minute))
	register("a", "c2", time.Now().Add(-time.Minute))

	// A machine's report on what it was sent, heard as the controller hears it.
	listener := nudgingListener{Enrolment: link.Enrolment{Inventory: inv}}
	seq := int64(0)
	reports := func(node string) {
		t.Helper()
		last, found, err := inv.SentTo(ctx, node)
		if err != nil || !found {
			t.Fatalf("%s was sent nothing: %v", node, err)
		}
		seq++
		body, _ := json.Marshal(map[string]any{"node": node, "applied": []string{"x.server"}, "declared": last.Digest,
			"report_sequence": seq})
		var report link.Report
		if err := json.Unmarshal(body, &report); err != nil {
			t.Fatal(err)
		}
		if _, err := listener.Heard(ctx, report); err != nil {
			t.Fatal(err)
		}
	}
	heldKeys := func() map[silentWait]time.Time {
		c := silentClockOf()
		if c == nil {
			return nil
		}
		held, err := c.Held(ctx)
		if err != nil {
			t.Fatal(err)
		}
		return held
	}
	plan := func() inventory.Plan {
		t.Helper()
		p, err := inv.PlanByID(ctx, "plan-974")
		if err != nil {
			t.Fatal(err)
		}
		return p
	}

	// The first controller. Its plan has tier 0 built and tier 1 not asked: nothing but the save asks anything.
	first, stopFirst := context.WithCancel(ctx)
	defer stopFirst()
	waitFirst := startRolloutEvents(first, open, js)
	built := time.Now().UTC()
	p := inventory.Plan{ID: "plan-974", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: built,
		State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
		Modules: map[string]*inventory.PlanModule{
			"a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"},
			"b": {},
		}}
	if err := inv.SavePlan(ctx, &p); err != nil {
		t.Fatal(err)
	}
	eventually(t, "tier 0 sent to both machines and tier 1 asked, on the save alone", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return len(sends) == 1 && len(*asked) == 1 && plan().Tier == 1
	})
	eventually(t, "a key on the bus for each machine tier 0 was sent to", func() bool { return len(heldKeys()) == 2 })

	// The controller is replaced: its looks and its watcher end. anchor reports meanwhile, to nobody.
	stopFirst()
	waitFirst()
	time.Sleep(300 * time.Millisecond)
	reports("anchor")

	second, stopSecond := context.WithCancel(ctx)
	defer stopSecond()
	waitSecond := startRolloutEvents(second, open, js)
	// Ends with the test before the cleanups restore what the look reads, which run after it (issue 1338).
	t.Cleanup(func() { stopSecond(); waitSecond() })
	// Its first look finds anchor reported, so its key goes; laptop's stays, and is the new controller's to hear expire.
	eventually(t, "the new controller's first look deletes the key of the machine that reported", func() bool {
		h := heldKeys()
		_, anchor := h[silentWait{Plan: "plan-974", Tier: 0, Node: "anchor"}]
		return !anchor
	})

	// tier 1's module is taken from its key by the new controller's look.
	eventually(t, "tier 1 sent to both machines", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return len(sends) == 2
	})
	reports("anchor")

	// laptop never reports: its key expires and the new controller names it silent, which ends the rollout.
	eventually(t, "the rollout ended, naming laptop silent and not anchor", func() bool {
		p := plan()
		return p.State == inventory.PlanDone
	})
	p = plan()
	if !strings.Contains(p.Note, "laptop") || strings.Contains(p.Note, "anchor") || !strings.Contains(p.Note, "said nothing") {
		t.Errorf("the plan ended with the note %q; want laptop named silent and not anchor", p.Note)
	}
	eventually(t, "no key left on the bus", func() bool { return len(heldKeys()) == 0 })

	// No tier asked twice and none skipped: one ask for tier 1 (tier 0 was built when the plan was kept), each tier sent
	// once to both machines, and said as sent once for each.
	mu.Lock()
	if len(*asked) != 1 || !strings.HasPrefix((*asked)[0], "novox/mesh-catalog#key-") {
		t.Errorf("taken from keys = %v; want tier 1's module taken once", *asked)
	}
	if len(sends) != 2 || strings.Join(sends[0], ",") != "anchor,laptop" || strings.Join(sends[1], ",") != "anchor,laptop" {
		t.Errorf("sends = %v; want both machines sent each tier once", sends)
	}
	mu.Unlock()
	said := map[string]int{}
	for deadline := time.After(5 * time.Second); len(said) < 4; {
		select {
		case m := <-sentHeard:
			var body planSentBody
			if err := json.Unmarshal(m.Data, &body); err != nil {
				t.Fatal(err)
			}
			said[fmt.Sprintf("%d/%s", body.Tier, body.Node)]++
		case <-deadline:
			t.Fatalf("sent was said for %v; want tiers 0 and 1 to both machines", said)
		}
	}
	time.Sleep(300 * time.Millisecond)
	for len(sentHeard) > 0 {
		var body planSentBody
		_ = json.Unmarshal((<-sentHeard).Data, &body)
		said[fmt.Sprintf("%d/%s", body.Tier, body.Node)]++
	}
	for k, n := range said {
		if n != 1 {
			t.Errorf("sent for %s was said %d times", k, n)
		}
	}
	// And every move of the plan was said as plan-moved.
	saved.noUnsaid(t, open)
}

// Acceptance of the verb path (issue 974, review 1), live, with the store and a bus: a walk the delivery's owner lets
// go by the verb `deliver` after the operator's approval is asked its first tier with no other event. The verb runs as a
// child process, whose saves the serving controller does not hear; the handler asks the look as the verb ends.
func TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := broker.AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	legacy, err := js.Conn().JetStream()
	if err != nil {
		t.Fatal(err)
	}
	checkEvents = link.OverNATS{Conn: js.Conn(), JS: legacy}
	t.Cleanup(func() { checkEvents = nil })
	asked := asksRecorded(t)
	register(t, open, catalogue.Manifest{Module: "app", Version: "1"})
	if _, err := assign(ctx, open, "anchor", "app"); err != nil {
		t.Fatal(err)
	}
	cut := time.Now().UTC()
	waiting := inventory.Plan{ID: "plan-974-deliver", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c8c8c8c8",
		Created: cut, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
		Modules:  map[string]*inventory.PlanModule{"app": {}},
		Delivery: &inventory.PlanDelivery{Awaits: catalogue.DeliverySeat},
		Times:    &inventory.PlanTimes{Cut: &cut, Class: inventory.ClassLeaf}}
	if err := open.inventory.SavePlan(ctx, &waiting); err != nil {
		t.Fatal(err)
	}

	// The serving controller, as serve starts it.
	wasSaved := inventory.PlanSaved
	inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open)
	t.Cleanup(func() { inventory.PlanSaved = wasSaved })
	started, stop := context.WithCancel(ctx)
	defer stop()
	wait := startRolloutEvents(started, open, js)
	time.Sleep(500 * time.Millisecond) // the first look: the walk waits, and is asked nothing

	// The verb as its child process runs it: the delivery command, whose saves are not the serving controller's.
	wasRun := runTheVerb
	runTheVerb = func(ctx context.Context, argv []string) (verbAnswer, error) {
		was := inventory.PlanSaved
		inventory.PlanSaved = func(*inventory.Inventory, inventory.Plan) {}
		defer func() { inventory.PlanSaved = was }()
		return verbAnswer{}, deliveryCommand(ctx, []string{"go", waiting.ID, "--why", "its turn"})
	}
	t.Cleanup(func() { runTheVerb = wasRun })
	// Registered after every variable cleanup above, so it runs before them (last in, first out): the look at the plans
	// has returned when checkEvents, inventory.PlanSaved and runTheVerb are restored (issue 1338).
	t.Cleanup(func() { stop(); wait() })
	handlers, _, err := seatToolHandlers()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := handlers["deliver"](ctx, []byte(`{"plan":"`+waiting.ID+`","why":"its turn"}`)); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the walk's first tier asked, with no event but the verb's end", func() bool {
		p, err := open.inventory.PlanByID(ctx, waiting.ID)
		return err == nil && p.Modules["app"] != nil && p.Modules["app"].Build != "" && len(*asked) == 1
	})
}