mesh-controller/cmd/mesh-controller/plan_looks_test.go

package main

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"

	"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"
)

func rolloutOf(sentAgo time.Duration, reported bool) (inventory.Plan, []inventory.Reported) {
	sent := time.Now().Add(-sentAgo).UTC()
	p := inventory.Plan{ID: "plan-9", State: inventory.PlanRolling, Tiers: [][]string{{"a", "b"}}, Tier: 0,
		Modules: map[string]*inventory.PlanModule{
			"a": {SentAt: &sent, Rest: map[string]inventory.SentDeclaration{"ace": {Digest: "d1", At: sent}}},
			"b": {SentAt: &sent, Rest: map[string]inventory.SentDeclaration{"ace": {Digest: "d1", At: sent}}},
		}}
	r := inventory.Reported{Node: "ace", Sent: &sent}
	if reported {
		at := sent.Add(time.Minute)
		r.At, r.Current, r.Declared, r.Outcome = &at, true, "d1", inventory.OutcomeApplied
	}
	return p, []inventory.Reported{r}
}

// A machine a plan sent and heard nothing from is a wait with a key, from when it was sent (issue 974, novox/hq ADR
// 0406 §5); one that reported on the send, or that is silent already, is none.
func TestAPlanWaitsOnTheMachinesItSentAndHeardNothingFrom(t *testing.T) {
	now := time.Now().UTC()
	p, reports := rolloutOf(5*time.Minute, false)
	waits := silentWaitsOf(p, reports, now, nil)
	if len(waits) != 1 {
		t.Fatalf("waits = %v; want the one machine", waits)
	}
	for w, at := range waits {
		if w.Plan != "plan-9" || w.Node != "ace" || w.Tier != 0 || now.Sub(at) < 4*time.Minute || now.Sub(at) > 6*time.Minute {
			t.Errorf("wait %+v sent at %v is wrong", w, at)
		}
	}
	p, reports = rolloutOf(5*time.Minute, true)
	if waits := silentWaitsOf(p, reports, now, nil); len(waits) != 0 {
		t.Errorf("a machine that reported on its send is waited on: %v", waits)
	}
	p, reports = rolloutOf(inventory.ApplySilentAfter+time.Minute, false)
	if waits := silentWaitsOf(p, reports, now, nil); len(waits) != 0 {
		t.Errorf("a machine silent past its time is waited on again: %v", waits)
	}
	p, reports = rolloutOf(time.Minute, false)
	p.State = inventory.PlanDone
	if waits := silentWaitsOf(p, reports, now, nil); len(waits) != 0 {
		t.Errorf("a plan that ended is waiting: %v", waits)
	}
}

// A release's open gate waits on its machines from when it began judging, and its verdict ends the wait.
func TestAReleasesOpenGateWaitsOnItsMachines(t *testing.T) {
	since := time.Now().Add(-time.Minute).UTC()
	g := &inventory.PlanGate{Machines: []string{"g14", "novox"}, Since: &since}
	p := inventory.Plan{ID: "release-1", State: inventory.PlanRolling, Release: &inventory.PlanRelease{Gate: g}}
	waits := silentWaitsOf(p, nil, time.Now(), nil)
	if len(waits) != 2 {
		t.Fatalf("waits = %v; want g14 and novox", waits)
	}
	g.Verdict = inventory.GatePassed
	if waits := silentWaitsOf(p, nil, time.Now(), nil); len(waits) != 0 {
		t.Errorf("a gate with its verdict still waits: %v", waits)
	}
}

// The keys follow the waits: a wait with no key, or a key of an earlier send, is armed; a key with no wait is deleted.
func TestTheKeysFollowTheWaits(t *testing.T) {
	t1 := time.Now().UTC()
	t2 := t1.Add(time.Minute)
	a, b, c := silentWait{"p", 0, "a"}, silentWait{"p", 0, "b"}, silentWait{"p", 0, "c"}
	arm, drop := silentDiff(map[silentWait]time.Time{a: t1, b: t2, c: t1}, map[silentWait]time.Time{a: t1, b: t1, silentWait{"p", 1, "a"}: t1})
	if len(arm) != 2 || !(arm[0] == b || arm[1] == b) || !(arm[0] == c || arm[1] == c) {
		t.Errorf("armed %v; want b (sent again) and c (no key)", arm)
	}
	if len(drop) != 1 || drop[0] != (silentWait{"p", 1, "a"}) {
		t.Errorf("dropped %v; want the key with no wait", drop)
	}
	if ttl := silentTTL(t1, t1.Add(inventory.ApplySilentAfter+time.Hour)); ttl != time.Second {
		t.Errorf("a wait long over has a key of %v; want the least, one second", ttl)
	}
}

func lookerCounting(failures *atomic.Int32, looks *atomic.Int32, block chan struct{}) *looker {
	l := newLooker(func(ctx context.Context) bool {
		looks.Add(1)
		if block != nil {
			<-block
		}
		return failures.Add(-1) >= 0
	})
	l.gap, l.base, l.most = 0, 20*time.Millisecond, 80*time.Millisecond
	return l
}

func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	for end := time.Now().Add(15 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) {
		if ok() {
			return
		}
	}
	t.Fatalf("never: %s", what)
}

// Asks made while a look runs are not lost and are one look after it, not one each.
func TestAsksMadeWhileALookRunsAreOneLookAfterIt(t *testing.T) {
	var failures, looks atomic.Int32
	block := make(chan struct{})
	l := lookerCounting(&failures, &looks, block)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	l.ask()
	eventually(t, "the first look starts", func() bool { return looks.Load() == 1 })
	for i := 0; i < 20; i++ {
		l.ask()
	}
	block <- struct{}{} // the first look ends
	eventually(t, "the asks made meanwhile are looked at", func() bool { return looks.Load() == 2 })
	block <- struct{}{}
	time.Sleep(60 * time.Millisecond)
	if n := looks.Load(); n != 2 {
		t.Errorf("%d looks for 21 asks; want 2", n)
	}
}

// A look that failed a step is looked at again, one at a time with a longer wait each time, and a look that fails
// nothing ends the retries (novox/hq ADR 0406 §5; the one named retry).
func TestAFailedLookIsLookedAtAgainAndAGoodOneEndsTheRetries(t *testing.T) {
	var failures, looks atomic.Int32
	failures.Store(3) // three looks fail, then one is good
	l := lookerCounting(&failures, &looks, nil)
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	l.ask()
	eventually(t, "four looks", func() bool { return looks.Load() == 4 })
	time.Sleep(200 * time.Millisecond)
	if n := looks.Load(); n != 4 {
		t.Errorf("%d looks; the good one should have ended the retries at 4", n)
	}
	l.mu.Lock()
	armed, backoff := l.retry != nil, l.backoff
	l.mu.Unlock()
	if armed || backoff != 0 {
		t.Errorf("a retry is armed (%v) or the wait kept (%v) after a good look", armed, backoff)
	}
}

func TestTheRetryWaitDoublesToItsMost(t *testing.T) {
	l := newLooker(func(context.Context) bool { return true })
	l.base, l.most = 5*time.Second, time.Minute
	var seen []time.Duration
	for i := 0; i < 6; i++ {
		l.after(true)
		seen = append(seen, l.backoff)
		l.retry.Stop()
		l.retry = nil
	}
	want := []time.Duration{5 * time.Second, 10 * time.Second, 20 * time.Second, 40 * time.Second, time.Minute, time.Minute}
	for i := range want {
		if seen[i] != want[i] {
			t.Fatalf("waits %v; want %v", seen, want)
		}
	}
}

// A plan saved asks for a look, and a walk saved done is armed to be recorded once its end can be known.
func TestAPlanSavedAsksForALookAndADoneWalkIsArmedToBeRecorded(t *testing.T) {
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)

	planKept(inventory.Plan{ID: "walk-1", State: inventory.PlanDone})
	t.Cleanup(func() {
		walkEnds.mu.Lock()
		defer walkEnds.mu.Unlock()
		if t := walkEnds.timers["walk-1"]; t != nil {
			t.Stop()
		}
		delete(walkEnds.timers, "walk-1")
	})
	eventually(t, "a look", func() bool { return looks.Load() == 1 })
	walkEnds.mu.Lock()
	_, armed := walkEnds.timers["walk-1"]
	walkEnds.mu.Unlock()
	if !armed {
		t.Error("a walk saved done is not armed to be recorded")
	}
	askToLook() // no panic, no looker, once it is taken away
}

// The tier-sends a step made are said once each, per machine, with the modules it was sent.
func TestTheSendsAStepMadeAreSaidPerMachine(t *testing.T) {
	sends := &sendsMade{}
	ctx := withSendsMade(context.Background(), sends)
	at := time.Now().UTC()
	noteSends(ctx, 2, []string{"ace"}, []string{"a", "b"}, at)
	noteSends(ctx, 2, []string{"g14"}, []string{"a"}, at)
	noteSends(context.Background(), 2, []string{"nobody"}, []string{"a"}, at) // outside a step: nothing kept
	got := sentBodies("plan-9", sends)
	if len(got) != 2 || got[0].Node != "ace" || len(got[0].Modules) != 2 || got[0].Tier != 2 || got[0].Plan != "plan-9" ||
		got[0].Event != "sent" || !got[0].At.Equal(at) || got[1].Node != "g14" {
		t.Errorf("sent bodies = %+v", got)
	}
}

// A silent key that expires asks for a look; a key a report deleted does not (issue 974, novox/hq ADR 0406 §5).
func TestAnExpiredSilentKeyAsksForALookAndAReportsDeleteDoesNot(t *testing.T) {
	js, err := broker.Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)
	if err := js.EnsureControllerBuckets(); err != nil {
		t.Fatal(err)
	}
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock(); setSilentClock(nil) })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	if err := startRolloutSilent(ctx, js); err != nil {
		t.Fatal(err)
	}
	c := silentClockOf()
	w := silentWait{Plan: "plan-5", Tier: 0, Node: "ace"}
	if err := c.Arm(ctx, w, time.Second, time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := c.Reported(ctx, w); err != nil {
		t.Fatal(err)
	}
	time.Sleep(1800 * time.Millisecond)
	if n := looks.Load(); n != 0 {
		t.Fatalf("a key a report deleted asked for %d looks", n)
	}
	if err := c.Arm(ctx, w, time.Second, time.Now()); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the expiry asks for a look", func() bool { return looks.Load() == 1 })
}

// A provisioner client's word on a module's files asks for a look, as a send that handed it modules is applied only
// once it has said so (issue 974, review 1): no walk waits for its silent key for what the client already said.
func TestAProvisionerClientsWordAsksForALook(t *testing.T) {
	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)
	}
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	if err := startProvAppliedLooks(ctx, js); err != nil {
		t.Fatal(err)
	}
	legacy, err := js.Conn().JetStream()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := legacy.Publish(broker.ProvisionerAppliedSubject("ace", "plex"), []byte(`{"status":"applied"}`)); err != nil {
		t.Fatal(err)
	}
	eventually(t, "a look on the client's word", func() bool { return looks.Load() >= 1 })
}

type brokenClock struct{ silentClock }

func (brokenClock) Held(context.Context) (map[silentWait]time.Time, error) {
	return nil, errors.New("the bus is down")
}

// A bus that will not say which keys it holds does not leave the waits unwatched: each is looked at by a timer of its own
// and the error is given back, so the look is marked failed and retried (issue 974, review 1).
func TestKeysThatCannotBeReadAreWatchedByTimersAndTheLookFails(t *testing.T) {
	w := silentWait{Plan: "plan-held", Tier: 0, Node: "ace"}
	sent := time.Now().UTC()
	t.Cleanup(func() { silentTimers.mu.Lock(); delete(silentTimers.timers, w); silentTimers.mu.Unlock() })
	err := keepSilentKeys(t.Context(), brokenClock{}, map[silentWait]time.Time{w: sent}, sent)
	if err == nil {
		t.Fatal("no error given back")
	}
	silentTimers.mu.Lock()
	armed := silentTimers.timers[w].Equal(sent)
	silentTimers.mu.Unlock()
	if !armed {
		t.Error("no timer of its own watches the wait")
	}
}

// A plan moved by a process the serving controller did not spawn — a verb run at the control node's terminal — is
// looked at: the controller follows its own plan-moved (issue 974, round 1).
func TestAPlanMovedByAnotherProcessAsksForALook(t *testing.T) {
	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)
	}
	var failures, looks atomic.Int32
	l := lookerCounting(&failures, &looks, nil)
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	t.Cleanup(func() { lookerMu.Lock(); theLooker = nil; lookerMu.Unlock() })
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go l.run(ctx)
	if err := startPlanMovedLooks(ctx, js); err != nil {
		t.Fatal(err)
	}
	legacy, err := js.Conn().JetStream()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := legacy.Publish(link.SeatEventSubject(link.MeshControllerSeat, link.KeyPlanMoved), []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	eventually(t, "a look on the plan moved", func() bool { return looks.Load() >= 1 })
}