mesh-controller/cmd/mesh-controller/away_integration_test.go

//go:build integration

package main

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

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

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

// 1161/AC5, with the store and a real bus and no ticker: a plan of two tiers whose first tier runs on a machine away when it
// is sent finishes that tier without waiting for it and without its bound, sends the second tier to the machine heard
// alone, and says the machine was away (issue 1161). The machine away keeps its want: it was sent nothing of the plan's
// builds, so the send it gets when it is heard again carries them (AC3).
func TestAPlanGoesOnWithoutAMachineAwayWhenItsTierIsSent(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	inv := open.inventory

	// laptop is away as every tier is sent; anchor is heard.
	wasAway := judgeAway
	judgeAway = func(context.Context, *stores, time.Time) (map[string]bool, error) {
		return map[string]bool{"laptop": true}, nil
	}
	t.Cleanup(func() { judgeAway = wasAway })

	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() })

	wasSaved := inventory.PlanSaved
	inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open)
	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 asks []string
	var sends [][]string
	wasAsk := askABuild
	askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) {
		mu.Lock()
		defer mu.Unlock()
		id := link.NewBuildID(time.Now().Add(time.Duration(len(asks)) * time.Millisecond))
		asks = append(asks, source.Repository+"#"+id)
		return id, nil
	}
	t.Cleanup(func() { askABuild = wasAsk })
	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)
		}
	}
	plan := func() inventory.Plan {
		t.Helper()
		p, err := inv.PlanByID(ctx, "plan-1160")
		if err != nil {
			t.Fatal(err)
		}
		return p
	}

	built := time.Now().UTC()
	p := inventory.Plan{ID: "plan-1160", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: time.Now().UTC(),
		State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
		Delivery: &inventory.PlanDelivery{Order: []inventory.Edge{{From: "b", To: "a", Kind: edgeChangeOrder}}},
		Modules: map[string]*inventory.PlanModule{
			"a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"},
			"b": {},
		}}
	first, stopFirst := context.WithCancel(ctx)
	defer stopFirst()
	startRolloutEvents(first, open, js)
	if err := inv.SavePlan(ctx, &p); err != nil {
		t.Fatal(err)
	}
	eventually(t, "tier 0 sent to anchor alone", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return len(sends) == 1 && strings.Join(sends[0], ",") == "anchor"
	})
	reports("anchor")
	// The note of the move to tier 1 is read in the step that sees it: later steps of the tier say their own.
	var moved string
	eventually(t, "tier 1 reached on anchor's report alone", func() bool {
		if q := plan(); q.Tier == 1 {
			moved = q.Note
			return true
		}
		return false
	})
	if !strings.Contains(moved, "laptop was away: sent when it is heard again") || strings.Contains(moved, "anchor") {
		t.Errorf("1161/AC2: the plan says %q; want laptop named away and not anchor", moved)
	}
	// The ask for tier 1's build is the next step, and records its build id on the module.
	eventually(t, "tier 1 asked", func() bool { m := plan().Modules["b"]; return m != nil && m.Build != "" })
	b := plan().Modules["b"]
	askedAt, _ := link.BuildAskedAt(b.Build)
	record(b.Build, "b", "c2", askedAt)
	register("b", "c2", askedAt)
	planBuilt(ctx, open, "b", "c2", "", askedAt, b.Build)
	eventually(t, "tier 1 sent to anchor alone", func() bool {
		mu.Lock()
		defer mu.Unlock()
		return len(sends) == 2 && strings.Join(sends[1], ",") == "anchor"
	})
	// 1161/AC3: laptop was sent nothing of the plan's builds, so it still wants them for the send it gets when heard again.
	was, known, err := inv.SentBuilds(ctx, "laptop")
	if err != nil || !known {
		t.Fatalf("1161/AC3: laptop's last send: %v, known %v", err, known)
	}
	current, err := inv.CurrentBuilds(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if was["a"] == current["a"].Commit || was["a"] != "c1" {
		t.Errorf("1161/AC3: laptop was last sent a at %q and a is now %q; want it still at c1, owed c2", was["a"], current["a"].Commit)
	}
	if got := plan().Away; strings.Join(got, ",") != "laptop" {
		t.Errorf("1161/AC2: the plan keeps %v away; want laptop", got)
	}
}