mesh-tools
013dad4688
Swaps three import paths for mesh-sdk's go/catalogue, go/overlay and go/validate (the same packages, copied from this repository at 6c2a8d62) and lets gofmt order the groups. The six swap commits compile together once the pin commit (go.mod, modules.txt) is on the branch; each alone does not, because a file on the SDK's types cannot call one still on the old package.
309 lines
12 KiB
Go
309 lines
12 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"github.com/nats-io/nats.go"
|
|
"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"
|
|
)
|
|
|
|
// 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())
|
|
}
|
|
}
|