450 lines
17 KiB
Go
450 lines
17 KiB
Go
//go:build integration
|
||
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"slices"
|
||
"strconv"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/nats-io/nats.go"
|
||
"github.com/novox/mesh-controller/internal/broker"
|
||
"github.com/novox/mesh-controller/internal/conditions"
|
||
"github.com/novox/mesh-controller/internal/inventory"
|
||
"github.com/novox/mesh-controller/internal/testbus"
|
||
)
|
||
|
||
func TestH1AsksAMachineToReportAndSendsItAgain(t *testing.T) {
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
h, told, clock := healingOn(t, open)
|
||
record, err := open.inventory.NodeByName(ctx, "laptop")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := open.inventory.RecordSent(ctx, record.ID, "d2", nil); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
key := sentNotReported(t, "laptop")
|
||
|
||
// First: the report had not reached the mesh, and asking for it brings it.
|
||
asked := 0
|
||
h.askReport = func(ctx context.Context, node string) error {
|
||
asked++
|
||
if node != "laptop" {
|
||
t.Errorf("asked %s", node)
|
||
}
|
||
_, err := open.inventory.RecordDoing(ctx, record.ID, inventory.Doing{Outcome: inventory.OutcomeApplied,
|
||
At: time.Now().Add(time.Second), Declared: "d2"})
|
||
return err
|
||
}
|
||
h.tick(ctx)
|
||
c, found, err := conditionsFrom.Get(ctx, key)
|
||
if err != nil || !found {
|
||
t.Fatalf("the condition is gone after the act — a healer cleared it, not an observation: %v", err)
|
||
}
|
||
if asked != 1 || len(c.Tried) != 1 || c.Tried[0].By != "healer H1" || !strings.Contains(c.Tried[0].Outcome, "acted") ||
|
||
c.Resolver != "healer:H1" {
|
||
t.Fatalf("after the first act: asked %d, %+v", asked, c)
|
||
}
|
||
if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H1" || acts[0].Outcome != inventory.HealActed ||
|
||
acts[0].Condition != key || acts[0].By != "healer H1" || acts[0].Epoch != 57 {
|
||
t.Fatalf("the act was not said as healer-acted: %+v", acts)
|
||
}
|
||
|
||
// Within its settle, nothing more: the observation has its turn.
|
||
clock.pass(time.Minute)
|
||
h.tick(ctx)
|
||
if asked != 1 {
|
||
t.Fatalf("acted again inside the settle: asked %d", asked)
|
||
}
|
||
|
||
// Second: the machine says nothing, so it is sent again.
|
||
clock.pass(3 * time.Minute)
|
||
sent := 0
|
||
h.askReport = func(context.Context, string) error { asked++; return nil }
|
||
h.sendAgain = func(_ context.Context, node string) error { sent++; return nil }
|
||
h.tick(ctx)
|
||
if asked != 2 || sent != 1 {
|
||
t.Fatalf("the second act: asked %d, sent %d", asked, sent)
|
||
}
|
||
c, _, _ = conditionsFrom.Get(ctx, key)
|
||
if len(c.Tried) != 2 || !strings.Contains(c.Tried[1].Outcome, "sent again") || !strings.Contains(c.Tried[1].Outcome, "act 2 of 2") {
|
||
t.Fatalf("tried %+v", c.Tried)
|
||
}
|
||
|
||
// The budget is spent: the operator's, urgent, said; and no healer touches it again.
|
||
clock.pass(4 * time.Minute)
|
||
h.tick(ctx)
|
||
c, _, _ = conditionsFrom.Get(ctx, key)
|
||
if !c.Escalated() || c.Severity != conditions.Urgent || len(c.Tried) != 3 ||
|
||
!strings.Contains(c.Tried[2].Outcome, "budget of 2") {
|
||
t.Fatalf("not handed to the operator: %+v", c)
|
||
}
|
||
if acts := told.said(); len(acts) != 3 || acts[2].Outcome != inventory.HealEscalated {
|
||
t.Fatalf("the escalation was not said: %+v", acts)
|
||
}
|
||
clock.pass(time.Hour)
|
||
h.tick(ctx)
|
||
if asked != 2 || sent != 1 || len(told.said()) != 3 {
|
||
t.Fatalf("a healer acted on a condition the operator holds: asked %d sent %d said %d", asked, sent, len(told.said()))
|
||
}
|
||
heals, err := open.inventory.HealsSince(ctx, time.Now().Add(-time.Hour))
|
||
if err != nil || len(heals) != 3 || heals[0].Outcome != inventory.HealActed || heals[1].Outcome != inventory.HealActed ||
|
||
heals[2].Outcome != inventory.HealEscalated || heals[0].Epoch != 57 {
|
||
t.Fatalf("the heals kept: %+v %v", heals, err)
|
||
}
|
||
// And a heal is not a hand act: what S15 counts never sees it.
|
||
for _, x := range heals {
|
||
if x.Healer == "" || strings.HasPrefix(x.Act, "hand-act") {
|
||
t.Errorf("a heal reads as a hand act: %+v", x)
|
||
}
|
||
}
|
||
}
|
||
|
||
func TestNoHealerActsWithoutTheLease(t *testing.T) {
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
h, told, _ := healingOn(t, open)
|
||
sentNotReported(t, "laptop")
|
||
h.epoch = func(context.Context) (uint64, error) { return 0, nil }
|
||
h.tick(ctx)
|
||
h.epoch = func(context.Context) (uint64, error) { return 0, errors.New("the lease is not held") }
|
||
h.tick(ctx)
|
||
h.epoch = func(context.Context) (uint64, error) { return 57, nil }
|
||
h.acting = func() bool { return false }
|
||
h.tick(ctx)
|
||
if len(told.said()) != 0 {
|
||
t.Fatalf("a healer acted without the lease: %+v", told.said())
|
||
}
|
||
}
|
||
|
||
func TestTheBrakeStopsEveryHealerAndSaysSo(t *testing.T) {
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
h, told, clock := healingOn(t, open)
|
||
for i := 0; i < healBrakeLimit; i++ {
|
||
if _, err := open.inventory.BeginHeal(ctx, inventory.Heal{Healer: "H3", ConditionKey: "seat.x.anchor.silent",
|
||
Kind: "holder-silent", Act: "asserted", Outcome: inventory.HealActed,
|
||
At: clock.now().Add(-time.Duration(healBrakeLimit-i) * time.Minute)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
sentNotReported(t, "laptop")
|
||
h.tick(ctx) // the fakes fail the test if anything acts
|
||
braked, found, err := conditionsFrom.Get(ctx, "mesh.healers.braked")
|
||
if err != nil || !found || braked.Severity != conditions.Urgent || !strings.Contains(braked.Evidence[0].Said, "H3 on seat.x.anchor.silent ×12") {
|
||
t.Fatalf("the brake was not said: %+v %v", braked, err)
|
||
}
|
||
if len(told.said()) != 0 {
|
||
t.Fatalf("a healer acted under the brake: %+v", told.said())
|
||
}
|
||
// Past the hour of the first act, still held: it lets go an hour after the last.
|
||
clock.pass(30 * time.Minute)
|
||
h.tick(ctx)
|
||
if _, held, _ := conditionsFrom.Get(ctx, "mesh.healers.braked"); !held {
|
||
t.Fatal("the brake let go before an hour had passed since the last act")
|
||
}
|
||
clock.pass(31 * time.Minute)
|
||
asked := 0
|
||
h.askReport = func(context.Context, string) error { asked++; return nil }
|
||
h.sendAgain = func(context.Context, string) error { return nil }
|
||
h.tick(ctx)
|
||
if _, held, _ := conditionsFrom.Get(ctx, "mesh.healers.braked"); held {
|
||
t.Fatal("the brake held an hour after the last act")
|
||
}
|
||
if asked != 1 {
|
||
t.Fatalf("the healers did not act again after the brake let go: asked %d", asked)
|
||
}
|
||
}
|
||
|
||
func TestH2ClosesAPlanAnotherHasTakenOver(t *testing.T) {
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
h, told, _ := healingOn(t, open)
|
||
created := time.Now().UTC().Add(-time.Hour)
|
||
older := inventory.Plan{ID: "plan-old", Repository: "novox/app", Branch: "main", Commit: "0ld0ld0", Created: created,
|
||
State: inventory.PlanBuilding, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "asked"}}}
|
||
waiting := inventory.Plan{ID: "plan-other", Repository: "novox/other", Branch: "main", Commit: "07he707", Created: created,
|
||
State: inventory.PlanBuilding, Tiers: [][]string{{"b"}}, Modules: map[string]*inventory.PlanModule{"b": {State: "asked"}}}
|
||
newer := inventory.Plan{ID: "plan-new", Repository: "novox/app", Branch: "main", Commit: "new0new", Created: created.Add(time.Minute),
|
||
State: inventory.PlanDone, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {State: "built"}}}
|
||
for _, p := range []*inventory.Plan{&older, &waiting, &newer} {
|
||
if err := open.inventory.SavePlan(ctx, p); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
for _, id := range []string{"plan-old", "plan-other"} {
|
||
if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopePlan, ID: id, Kind: "stalled",
|
||
Severity: conditions.Warning, Summary: id + " is stalled", Source: "S3"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
h.tick(ctx)
|
||
closed, err := open.inventory.PlanByID(ctx, "plan-old")
|
||
if err != nil || closed.State != inventory.PlanSuperseded || !strings.Contains(closed.Note, "closed by healer H2") ||
|
||
!strings.Contains(closed.Note, "plan-new") {
|
||
t.Fatalf("the superseded plan: %+v %v", closed, err)
|
||
}
|
||
if other, _ := open.inventory.PlanByID(ctx, "plan-other"); other.State != inventory.PlanBuilding {
|
||
t.Fatalf("a plan still waiting was closed: %+v", other)
|
||
}
|
||
if c, _, _ := conditionsFrom.Get(ctx, "plan.plan-other.stalled"); len(c.Tried) != 0 || c.Resolver != conditions.ResolverSelf {
|
||
t.Fatalf("a plan H2 does not repair was touched: %+v", c)
|
||
}
|
||
if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H2" || acts[0].Condition != "plan.plan-old.stalled" {
|
||
t.Fatalf("said %+v", acts)
|
||
}
|
||
// Finished: every module built, none rolled out unsent — closed as done.
|
||
done := inventory.Plan{ID: "plan-done", Repository: "novox/third", Branch: "main", Commit: "d0ned0n", Created: created,
|
||
State: inventory.PlanRolling, Tier: 0, Tiers: [][]string{{"c"}}, Modules: map[string]*inventory.PlanModule{"c": {State: "built"}}}
|
||
if err := open.inventory.SavePlan(ctx, &done); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if state, why, err := planStale(ctx, open.inventory, done); err != nil || state != inventory.PlanDone || !strings.Contains(why, "finished") {
|
||
t.Fatalf("a finished plan reads %q %q %v", state, why, err)
|
||
}
|
||
}
|
||
|
||
func TestH4ResetsOnlyWhatTheTableMarksResettable(t *testing.T) {
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
h, _, _ := healingOn(t, open)
|
||
marked := 0
|
||
for _, c := range broker.MeshConsumers() {
|
||
if c.Resettable != "" {
|
||
marked++
|
||
if c.Stream != broker.EventsStream || c.Name != broker.ControllerName {
|
||
t.Errorf("%s on %s is marked resettable: only the controller's own events consumer is", c.Name, c.Stream)
|
||
}
|
||
}
|
||
}
|
||
if marked != 1 {
|
||
t.Fatalf("%d consumers are marked resettable, want the controller's events consumer alone", marked)
|
||
}
|
||
for _, who := range []string{"EVENTS.anchor_shop", "CONTROL.controller"} {
|
||
if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopeBus, ID: who, Token: "behind",
|
||
Kind: kindConsumerBehind, Severity: conditions.Warning, Summary: who + " is far behind", Source: "D6"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
h.tick(ctx) // the fake reset fails the test if it is called
|
||
reset := ""
|
||
h.resetConsumer = func(stream, name string) (string, error) {
|
||
reset = stream + "." + name
|
||
return "it was 1500 behind", nil
|
||
}
|
||
if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopeBus, ID: "EVENTS.controller",
|
||
Token: "behind", Kind: kindConsumerBehind, Severity: conditions.Warning, Summary: "far behind", Source: "D6"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
h.tick(ctx)
|
||
if reset != "EVENTS.controller" {
|
||
t.Fatalf("reset %q", reset)
|
||
}
|
||
}
|
||
|
||
func TestNatsH3AssertsAMissingConsumerAgainAndBrakesAfterItsBudget(t *testing.T) {
|
||
url := testbus.URL(t)
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
js, err := broker.Dial(url)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(js.Close)
|
||
for _, s := range []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS"} {
|
||
_ = js.Context().DeleteStream(s)
|
||
}
|
||
if _, err := assertBusObjects(ctx, open.inventory, js); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
h, told, clock := healingOn(t, open)
|
||
h.js = js
|
||
h.assertObjects = func(ctx context.Context) error { return assertOnSend(ctx, open.inventory, js, " ") }
|
||
d := &doctor{open: open, js: js}
|
||
probe := func() {
|
||
t.Helper()
|
||
found, err := probeConsumers(ctx, d)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := conditionsFrom.Reconcile(ctx, "D6", kindedAs(found, "consumer-wrong")); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
const key = "bus.NODES.laptop.missing"
|
||
|
||
if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
probe()
|
||
if c, raised, _ := conditionsFrom.Get(ctx, key); !raised || c.Kind != "consumer-lost" {
|
||
t.Fatalf("the deleted consumer was not raised: %+v", c)
|
||
}
|
||
h.tick(ctx)
|
||
if _, err := js.Context().ConsumerInfo("NODES", "laptop"); err != nil {
|
||
t.Fatalf("H3 did not make the consumer again: %v", err)
|
||
}
|
||
c, still, _ := conditionsFrom.Get(ctx, key)
|
||
if !still || len(c.Tried) != 1 || c.Tried[0].By != "healer H3" || c.Resolver != "healer:H3" {
|
||
t.Fatalf("the act is not in the condition, or the healer cleared it: %+v", c)
|
||
}
|
||
probe()
|
||
if _, still, _ := conditionsFrom.Get(ctx, key); still {
|
||
t.Fatal("the probe's next run did not clear what H3 repaired")
|
||
}
|
||
// Kept in the history as the keeper says it, a moment later.
|
||
var cleared *conditions.Event
|
||
for wait := time.Now().Add(5 * time.Second); cleared == nil && time.Now().Before(wait); time.Sleep(20 * time.Millisecond) {
|
||
history, err := conditionsFrom.HistorySince(ctx, time.Now().Add(-time.Minute))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for i := range history {
|
||
if history[i].Key == key && history[i].Change == conditions.ChangeCleared {
|
||
cleared = &history[i]
|
||
}
|
||
}
|
||
}
|
||
if cleared == nil || !strings.Contains(cleared.Why, "D6 no longer observes it") || len(cleared.Tried) != 1 {
|
||
t.Fatalf("the clearing is not the probe's, or forgets what was tried: %+v", cleared)
|
||
}
|
||
|
||
// Again within the hour: the budget is one, so after its settle the operator is told, and H3 stops.
|
||
clock.pass(10 * time.Minute)
|
||
if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
probe()
|
||
h.assertObjects = func(context.Context) error { t.Error("H3 acted past its budget"); return nil }
|
||
h.tick(ctx)
|
||
c, _, _ = conditionsFrom.Get(ctx, key)
|
||
if !c.Escalated() || c.Severity != conditions.Urgent || c.Count != 2 {
|
||
t.Fatalf("the spent budget was not handed to the operator: %+v", c)
|
||
}
|
||
acts := told.said()
|
||
if len(acts) != 2 || acts[0].Outcome != inventory.HealActed || acts[1].Outcome != inventory.HealEscalated {
|
||
t.Fatalf("said %+v", acts)
|
||
}
|
||
clock.pass(2 * time.Hour)
|
||
h.tick(ctx)
|
||
if len(told.said()) != 2 {
|
||
t.Fatal("a healer acted on what the operator holds")
|
||
}
|
||
}
|
||
|
||
func TestNatsH4ResetsTheControllersEventsConsumerAndItStillDelivers(t *testing.T) {
|
||
url := testbus.URL(t)
|
||
open := aMesh(t)
|
||
ctx := t.Context()
|
||
js, err := broker.Dial(url)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(js.Close)
|
||
for _, s := range []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS"} {
|
||
_ = js.Context().DeleteStream(s)
|
||
}
|
||
if _, err := assertBusObjects(ctx, open.inventory, js); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// Bound as the controller binds it (link/receive_nats.go), taking one and acknowledging none.
|
||
events := make(chan *nats.Msg, 64)
|
||
sub, err := js.Context().ChanSubscribe("", events, nats.Bind(broker.EventsStream, broker.ControllerName))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||
followed := broker.ControllerFollows[0]
|
||
for i := 0; i < consumerFarBehind+200; i++ {
|
||
if _, err := js.Context().Publish(followed, []byte(`{"n":`+strconv.Itoa(i)+`}`)); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
d := &doctor{open: open, js: js}
|
||
probe := func() []conditions.Observation {
|
||
t.Helper()
|
||
found, err := probeConsumers(ctx, d)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := conditionsFrom.Reconcile(ctx, "D6", kindedAs(found, "consumer-wrong")); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return found
|
||
}
|
||
probe()
|
||
const key = "bus.EVENTS.controller.behind"
|
||
if c, raised, _ := conditionsFrom.Get(ctx, key); !raised || c.Kind != kindConsumerBehind {
|
||
t.Fatalf("a consumer %d behind was not raised: %+v", consumerFarBehind+200, c)
|
||
}
|
||
h, told, _ := healingOn(t, open)
|
||
h.resetConsumer = func(stream, name string) (string, error) {
|
||
before, after, err := js.ResetConsumer(stream, name)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return "it was " + strconv.FormatUint(before.Pending, 10) + " behind; " + strconv.FormatUint(after.Pending, 10) +
|
||
" pending now", nil
|
||
}
|
||
h.tick(ctx)
|
||
if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H4" || acts[0].Outcome != inventory.HealActed {
|
||
t.Fatalf("said %+v", acts)
|
||
}
|
||
if found := probe(); len(found) != 0 {
|
||
t.Fatalf("after the reset the probe still finds %+v", found)
|
||
}
|
||
if _, still, _ := conditionsFrom.Get(ctx, key); still {
|
||
t.Fatal("the probe's next run did not clear what H4 repaired")
|
||
}
|
||
// What comes next still reaches the controller.
|
||
for len(events) > 0 {
|
||
<-events
|
||
}
|
||
if _, err := js.Context().Publish(followed, []byte(`{"after":"the reset"}`)); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
deadline := time.After(10 * time.Second)
|
||
for {
|
||
select {
|
||
case m := <-events:
|
||
if strings.Contains(string(m.Data), "after") {
|
||
return
|
||
}
|
||
_ = m.Ack()
|
||
case <-deadline:
|
||
t.Fatal("the controller's subscription heard nothing after its consumer was reset: it would be deaf")
|
||
}
|
||
}
|
||
}
|
||
|
||
func TestNatsAskToReportReachesTheMachine(t *testing.T) {
|
||
url := testbus.URL(t)
|
||
conn, err := nats.Connect(url)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(conn.Close)
|
||
asked, err := conn.SubscribeSync(broker.AskReportSubject("laptop"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := askToReport(t.Context(), conn, "laptop"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
msg, err := asked.NextMsg(5 * time.Second)
|
||
if err != nil || !strings.Contains(string(msg.Data), "healer H1") {
|
||
t.Fatalf("the machine heard %v, %v", msg, err)
|
||
}
|
||
perms, err := broker.PermissionsFor(broker.Principal{Kind: broker.KindNode, Node: "laptop", PasswordHash: "x"})
|
||
if err != nil || !slices.Contains(perms.Subscribe, "mesh.node.laptop.ask.report") ||
|
||
slices.Contains(perms.Subscribe, "mesh.node.anchor.ask.report") {
|
||
t.Fatalf("a machine's grant for the question: %v %v", perms.Subscribe, err)
|
||
}
|
||
}
|