mesh-controller/cmd/mesh-controller/backups_before_send_integration_test.go

//go:build integration

package main

import (
	"encoding/json"
	"os"
	"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"
)

// aBackupSeatOnTheBus is one machine's node-backup holder as a test makes it: it counts the `now` asks it hears and
// answers `backed-up` with the last night it is told (novox/hq ADR 0404).
type aBackupSeatOnTheBus struct {
	mu    sync.Mutex
	nows  []string
	night map[string]any // module → lastNight
}

func (f *aBackupSeatOnTheBus) serve(t *testing.T, conn *nats.Conn, node string) {
	t.Helper()
	reply := func(m *nats.Msg, result any) {
		body, _ := json.Marshal(map[string]any{"result": result, "error": "", "node": node})
		_ = m.Respond(body)
	}
	now, err := conn.Subscribe(link.NodeSeatToolSubject(catalogue.BackupSeat, "now", node), func(m *nats.Msg) {
		var a struct {
			Module string `json:"module"`
		}
		_ = json.Unmarshal(m.Data, &a)
		f.mu.Lock()
		f.nows = append(f.nows, a.Module)
		f.mu.Unlock()
		reply(m, map[string]any{"started": a.Module})
	})
	if err != nil {
		t.Fatal(err)
	}
	read, err := conn.Subscribe(link.NodeSeatToolSubject(catalogue.BackupSeat, "backed-up", node), func(m *nats.Msg) {
		f.mu.Lock()
		defer f.mu.Unlock()
		var out []map[string]any
		for module, n := range f.night {
			out = append(out, map[string]any{"module": module, "lastNight": n})
		}
		reply(m, out)
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = now.Unsubscribe(); _ = read.Unsubscribe() })
	if err := conn.Flush(); err != nil {
		t.Fatal(err)
	}
}

func (f *aBackupSeatOnTheBus) answers(module string, ok bool, words string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.night = map[string]any{module: map[string]any{"ok": ok, "at": time.Now().UTC().Add(time.Second), "error": words}}
}

func (f *aBackupSeatOnTheBus) asks() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.nows)
}

// Acceptance test 3 of design 47, on the controller's side: a delivery plan that moves a module keeping data waits
// for its backups on both machines before it asks a build; a backup that fails holds it, with the machine, the
// module and the backup's words, and a condition says the same; the verb the delivery's release calls asks the
// backups again, and the plan goes on. One bus and one store for the scenarios.
func TestADeliveryThatMovesABackedUpModuleWaitsForItsBackupAndAFailedOneHoldsItAndIsSaid(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	inv := open.inventory
	asked := asksRecorded(t)
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	before := handActConn
	handActConn = js.Conn()
	t.Cleanup(func() { handActConn = before })

	if _, err := inv.SeedSeats(ctx, catalogue.DefaultSeats()); err != nil {
		t.Fatal(err)
	}
	register(t, open, catalogue.Manifest{Module: "keeper", Version: "1",
		Claims: []catalogue.Claim{{Name: catalogue.BackupSeat, Scope: catalogue.ScopeNode, Serves: []string{"backed-up", "now", "restore"}}}})
	notes, err := catalogue.ParseManifest([]byte(`{"module":"notes","version":"1","data":{"own":[` +
		`{"id":"files","path":"${dir:files}","class":"valuable","why":"people's notes"}]},` +
		`"resources":[{"id":"files","type":"directory"}]}`))
	if err != nil {
		t.Fatal(err)
	}
	if err := inv.RegisterModule(ctx, notes, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git",
		Path: "modules/notes", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
		t.Fatal(err)
	}
	for _, node := range []string{"anchor", "laptop"} {
		if _, err := assign(ctx, open, node, "keeper"); err != nil {
			t.Fatal(err)
		}
		if _, err := assign(ctx, open, node, "notes"); err != nil {
			t.Fatal(err)
		}
	}
	seats := map[string]*aBackupSeatOnTheBus{"anchor": {}, "laptop": {}}
	for node, s := range seats {
		s.serve(t, js.Conn(), node)
	}
	// Short waits: the bound and the look are variables, so nothing here waits for the minutes of a real backup.
	wasWithin, wasEvery := backupsWithin, backupsPollEvery
	backupsWithin, backupsPollEvery = time.Minute, 10*time.Millisecond
	t.Cleanup(func() { backupsWithin, backupsPollEvery = wasWithin, wasEvery })

	newPlan := func(id string) {
		t.Helper()
		p := inventory.Plan{ID: id, Repository: "novox/mesh-catalog", Branch: "main", Commit: "c1", Created: time.Now().UTC(),
			State: inventory.PlanBuilding, Tiers: [][]string{{"notes"}}, Modules: map[string]*inventory.PlanModule{"notes": {}}}
		if err := inv.SavePlan(ctx, &p); err != nil {
			t.Fatal(err)
		}
	}
	read := func(id string) inventory.Plan {
		t.Helper()
		p, err := inv.PlanByID(ctx, id)
		if err != nil {
			t.Fatal(err)
		}
		return p
	}

	// 1. Both machines are asked; while one has not answered, no build is asked.
	newPlan("plan-one")
	advancePlans(ctx, open)
	if seats["anchor"].asks() != 1 || seats["laptop"].asks() != 1 || len(*asked) != 0 {
		t.Fatalf("asks anchor=%d laptop=%d builds=%v", seats["anchor"].asks(), seats["laptop"].asks(), *asked)
	}
	seats["anchor"].answers("notes", true, "")
	advancePlans(ctx, open)
	if len(*asked) != 0 || read("plan-one").Backups.State != inventory.BackupPending {
		t.Fatalf("a tier was asked while a backup was pending: %v %+v", *asked, read("plan-one").Backups)
	}
	seats["laptop"].answers("notes", true, "")
	advancePlans(ctx, open)
	if len(*asked) != 1 || read("plan-one").Backups.State != inventory.BackupDone {
		t.Fatalf("the tier was not asked once every backup was done: %v %+v", *asked, read("plan-one").Backups)
	}
	finish := read("plan-one")
	finish.State = inventory.PlanDone
	if err := inv.SavePlan(ctx, &finish); err != nil {
		t.Fatal(err)
	}

	// 2. A backup that fails holds the plan: not failed, nothing asked, the words kept, a condition raised.
	seats["anchor"].night, seats["laptop"].night = nil, nil
	newPlan("plan-two")
	advancePlans(ctx, open)
	seats["anchor"].answers("notes", true, "")
	seats["laptop"].answers("notes", false, "no space left on device")
	advancePlans(ctx, open)
	held := read("plan-two")
	if held.State != inventory.PlanBuilding || held.Backups.State != inventory.BackupFailed || len(*asked) != 1 {
		t.Fatalf("the plan was not held: %s %+v builds=%v", held.State, held.Backups, *asked)
	}
	var failed *inventory.PlanBackup
	for i := range held.Backups.Asked {
		if held.Backups.Asked[i].State == inventory.BackupFailed {
			failed = &held.Backups.Asked[i]
		}
	}
	if failed == nil || failed.Machine != "laptop" || failed.Module != "notes" || failed.Words != "no space left on device" ||
		strings.Contains(held.Note, "tried again") {
		t.Fatalf("the failed backup: %+v note=%q", failed, held.Note)
	}
	advancePlans(ctx, open) // not retried on the next look
	if again := read("plan-two"); again.Backups.State != inventory.BackupFailed || len(*asked) != 1 ||
		seats["laptop"].asks() != 2 {
		t.Fatalf("a held plan was asked again by itself: %+v builds=%v asks=%d", again.Backups, *asked, seats["laptop"].asks())
	}
	said, err := probeBackupHeld(ctx, &doctor{open: open})
	if err != nil || len(said) != 1 || said[0].Kind != kindBackupHeld || !strings.Contains(said[0].Said, "no space left on device") {
		t.Fatalf("the condition: %+v %v", said, err)
	}

	// 3. The verb the release calls: the failed backup is asked again, and only it, and the plan goes on.
	seats["laptop"].answers("notes", false, "no space left on device")
	if err := deliveryCommand(ctx, []string{"backups-again", "plan-two", "--why", "a person released it", "--by",
		"mesh-delivery for jochen"}); err != nil {
		t.Fatal(err)
	}
	if err := deliveryCommand(ctx, []string{"backups-again", "plan-two", "--why", "again"}); err == nil {
		t.Fatal("a plan with no failed backup was asked again")
	}
	seats["laptop"].night = nil
	advancePlans(ctx, open)
	if seats["laptop"].asks() != 3 || seats["anchor"].asks() != 2 {
		t.Fatalf("asked again: laptop=%d anchor=%d", seats["laptop"].asks(), seats["anchor"].asks())
	}
	seats["laptop"].answers("notes", true, "")
	advancePlans(ctx, open)
	if got := read("plan-two"); got.Backups.State != inventory.BackupDone || len(*asked) != 2 {
		t.Fatalf("the plan did not go on: %+v builds=%v", got.Backups, *asked)
	}
	if said, _ := probeBackupHeld(ctx, &doctor{open: open}); len(said) != 0 {
		t.Fatalf("the condition stays: %+v", said)
	}
}

// Issue 1086: a plan cut from a merge asked its first tier at the cut, so a module keeping data was sent without the
// backup the plan promised. The cut now takes the backups first (askFirstTierAtCut, batches.go), and the walk's own
// step asks the tier once they are done.
func TestAPlanCutFromAMergeOfABackedUpModuleAsksItsBackupAndNotItsTierUntilItIsDone(t *testing.T) {
	open := aMesh(t)
	ctx := t.Context()
	inv := open.inventory
	settings := t.TempDir() + "/merge-window.json"
	if err := os.WriteFile(settings, []byte(`{"merge-window": "90s", "merge-window-at-most": "10m"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	t.Setenv(mergeWindowFileVar, settings)
	everyHeadWaits(t)
	asked := asksRecorded(t)
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	before := handActConn
	handActConn = js.Conn()
	t.Cleanup(func() { handActConn = before })

	if _, err := inv.SeedSeats(ctx, catalogue.DefaultSeats()); err != nil {
		t.Fatal(err)
	}
	register(t, open, catalogue.Manifest{Module: "keeper", Version: "1",
		Claims: []catalogue.Claim{{Name: catalogue.BackupSeat, Scope: catalogue.ScopeNode, Serves: []string{"backed-up", "now", "restore"}}}})
	if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-controller", Version: "1"},
		inventory.Source{Repository: "novox/mesh-controller", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
		t.Fatal(err)
	}
	notes, err := catalogue.ParseManifest([]byte(`{"module":"notes","version":"1","data":{"own":[` +
		`{"id":"files","path":"${dir:files}","class":"valuable","why":"people's notes"}]},` +
		`"resources":[{"id":"files","type":"directory"}]}`))
	if err != nil {
		t.Fatal(err)
	}
	if err := inv.RegisterModule(ctx, notes, inventory.Source{Repository: "novox/notes", Seat: "git", Ref: "main",
		BuiltFrom: "c0", Head: "c0"}); err != nil {
		t.Fatal(err)
	}
	for _, m := range []string{"keeper", "notes"} {
		if _, err := assign(ctx, open, "anchor", m); err != nil {
			t.Fatal(err)
		}
	}
	seat := &aBackupSeatOnTheBus{}
	seat.serve(t, js.Conn(), "anchor")
	wasWithin, wasEvery := backupsWithin, backupsPollEvery
	backupsWithin, backupsPollEvery = time.Minute, 10*time.Millisecond
	t.Cleanup(func() { backupsWithin, backupsPollEvery = wasWithin, wasEvery })

	// 1. A merge of notes is heard and its window closes: the batch is cut through cutBatch.
	hear(t, open, repoMerge("notes", "c-notes", t0), t0.Add(time.Second))
	cutAt(t, open, t0.Add(time.Second+90*time.Second))
	ws, _ := walks(t, open)
	if len(ws) != 1 {
		t.Fatalf("%d walk(s)", len(ws))
	}
	// 2. Right after the cut no tier is asked, the record has anchor/notes pending, the seat heard one `now`.
	cut, err := inv.PlanByID(ctx, ws[0].ID)
	if err != nil {
		t.Fatal(err)
	}
	if len(*asked) != 0 || askedInTier(cut) {
		t.Fatalf("the first tier was asked at the cut: %v %+v", *asked, cut.Modules)
	}
	if cut.Backups == nil || cut.Backups.State != inventory.BackupPending || len(cut.Backups.Asked) != 1 ||
		cut.Backups.Asked[0].Machine != "anchor" || cut.Backups.Asked[0].Module != "notes" {
		t.Fatalf("the backups of the cut: %+v", cut.Backups)
	}
	if seat.asks() != 1 {
		t.Fatalf("the seat heard %d `now`", seat.asks())
	}
	// 3. The backup answers done: the walk's own step asks the first tier, once, and asks no second backup.
	seat.answers("notes", true, "")
	advancePlans(ctx, open)
	advancePlans(ctx, open)
	after, err := inv.PlanByID(ctx, ws[0].ID)
	if err != nil {
		t.Fatal(err)
	}
	if len(*asked) != 1 || after.Backups.State != inventory.BackupDone || seat.asks() != 1 {
		t.Fatalf("builds=%v backups=%+v asks=%d", *asked, after.Backups, seat.asks())
	}
}