issue 974: the step's acceptance test: a rollout advances on events with no ticker, a silent machine is named when its key expires, a replaced controller resumes

cmd/mesh-controller/plan_looks.go

@@ -156,6 +156,10 @@ func startPlanLooks(ctx context.Context, open *stores) {
 	l.ask()
 }
 
+// applySilentAfter is how long a machine may say nothing on a send before the plan names it silent: the mesh's
+// ApplySilentAfter, a variable so a test of the wait needs no quarter of an hour.
+var applySilentAfter = inventory.ApplySilentAfter
+
 // planLookPasses is the most times one look goes over the plans again because the first pass changed which are open.
 const planLookPasses = 6
 
@@ -316,7 +320,7 @@ func silentWaitsOf(p inventory.Plan, reports []inventory.Reported, now time.Time
 // silentTTL is how long a wait's key lives: what is left of ApplySilentAfter from the send, and a second over, so the
 // look it wakes finds the machine silent and not a fraction early.
 func silentTTL(sent, now time.Time) time.Duration {
-	left := sent.Add(inventory.ApplySilentAfter).Sub(now) + time.Second
+	left := sent.Add(applySilentAfter).Sub(now) + time.Second
 	if left < time.Second {
 		return time.Second
 	}
@@ -429,7 +433,11 @@ func startRolloutSilent(ctx context.Context, js *broker.JetStream) error {
 	setSilentClock(clock)
 	go func() {
 		<-ctx.Done()
-		setSilentClock(nil)
+		silentMu.Lock()
+		if silentKeyed == silentClock(clock) {
+			silentKeyed = nil
+		}
+		silentMu.Unlock()
 	}()
 	return nil
 }

cmd/mesh-controller/plan_looks_test.go

@@ -101,7 +101,7 @@ func lookerCounting(failures *atomic.Int32, looks *atomic.Int32, block chan stru
 
 func eventually(t *testing.T, what string, ok func() bool) {
 	t.Helper()
-	for end := time.Now().Add(3 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) {
+	for end := time.Now().Add(15 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) {
 		if ok() {
 			return
 		}

cmd/mesh-controller/release.go

@@ -1023,12 +1023,12 @@ func advanceRest(ctx context.Context, open *stores, p *inventory.Plan, now time.
 			outcome, reported = provisionedOutcome(node, sent.Only(carried...), byNode[node], reported, provLookup(ctx))
 		}
 		if !reported {
-			if now.Sub(*g.Since) > inventory.ApplySilentAfter {
+			if now.Sub(*g.Since) > applySilentAfter {
 				g.Silent = true
 				r.Silent = append(r.Silent, node)
 				changed = true
 				notes = append(notes, fmt.Sprintf("%s said nothing on its send within %s: left silent", node,
-					inventory.ApplySilentAfter))
+					applySilentAfter))
 				continue
 			}
 			judging = append(judging, node+": not yet reported on its send")

cmd/mesh-controller/release_plan.go

@@ -1174,7 +1174,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 			said = append(said, "failed or refused its send (apply-failing says it): "+strings.Join(failed, ", "))
 		}
 		if len(silent) > 0 {
-			said = append(said, fmt.Sprintf("said nothing on its send within %s: %s", inventory.ApplySilentAfter,
+			said = append(said, fmt.Sprintf("said nothing on its send within %s: %s", applySilentAfter,
 				strings.Join(silent, ", ")))
 		}
 		said = append(said, recordedSaid(*p)...)
@@ -1239,7 +1239,7 @@ func walkReports(p inventory.Plan, reports []inventory.Reported, now time.Time,
 				case strings.HasPrefix(outcome, inventory.OutcomeProvisioningFailed):
 					failed = append(failed, node+" ("+outcome+")")
 				}
-			case !at.IsZero() && now.Sub(at) > inventory.ApplySilentAfter:
+			case !at.IsZero() && now.Sub(at) > applySilentAfter:
 				seen[node] = true
 				silent = append(silent, node)
 			default:

cmd/mesh-controller/rollout_events_test.go

@@ -0,0 +1,281 @@
+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"
+)
+
+// Acceptance of the step (novox/hq ADR 0406 §5, §6, acceptance 7; issue 974), live, with the store and a real bus and no
+// ticker: a rollout of two tiers advances on the events alone — the plan kept, the machines' reports, a silent key's
+// expiry — a machine that reports in time is never named silent and one that never reports is named silent when its
+// key expires, and a controller replaced in the middle of it resumes with no tier asked twice and none skipped.
+func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
+	open := aMesh(t)
+	ctx := t.Context()
+	inv := open.inventory
+
+	// A quarter of an hour of silence is two seconds here.
+	wasSilent := applySilentAfter
+	applySilentAfter = 2 * time.Second
+	t.Cleanup(func() { applySilentAfter = wasSilent })
+
+	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() })
+
+	// Every save of a plan is said and asks the look, as the serving controller's hook does.
+	saved := &movesSaid{revs: map[string][]int64{}}
+	wasSaved := inventory.PlanSaved
+	inventory.PlanSaved = func(i *inventory.Inventory, p inventory.Plan) { saved.hear(i, p); planKept(p) }
+	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)
+		}
+	}
+	heldKeys := func() map[silentWait]time.Time {
+		c := silentClockOf()
+		if c == nil {
+			return nil
+		}
+		held, err := c.Held(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return held
+	}
+	plan := func() inventory.Plan {
+		t.Helper()
+		p, err := inv.PlanByID(ctx, "plan-974")
+		if err != nil {
+			t.Fatal(err)
+		}
+		return p
+	}
+
+	// The first controller. Its plan has tier 0 built and tier 1 not asked: nothing but the save asks anything.
+	first, stopFirst := context.WithCancel(ctx)
+	defer stopFirst()
+	if err := startRolloutSilent(first, js); err != nil {
+		t.Fatal(err)
+	}
+	startPlanLooks(first, open)
+	built := time.Now().UTC()
+	p := inventory.Plan{ID: "plan-974", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: built,
+		State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
+		Modules: map[string]*inventory.PlanModule{
+			"a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"},
+			"b": {},
+		}}
+	if err := inv.SavePlan(ctx, &p); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "tier 0 sent to both machines and tier 1 asked, on the save alone", func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(sends) == 1 && len(asks) == 1 && plan().Tier == 1
+	})
+	eventually(t, "a key on the bus for each machine tier 0 was sent to", func() bool { return len(heldKeys()) == 2 })
+
+	// The controller is replaced: its looks and its watcher end. anchor reports meanwhile, to nobody.
+	stopFirst()
+	time.Sleep(300 * time.Millisecond)
+	reports("anchor")
+
+	second, stopSecond := context.WithCancel(ctx)
+	defer stopSecond()
+	if err := startRolloutSilent(second, js); err != nil {
+		t.Fatal(err)
+	}
+	startPlanLooks(second, open)
+	// Its first look finds anchor reported, so its key goes; laptop's stays, and is the new controller's to hear expire.
+	eventually(t, "the new controller's first look deletes the key of the machine that reported", func() bool {
+		h := heldKeys()
+		_, anchor := h[silentWait{Plan: "plan-974", Tier: 0, Node: "anchor"}]
+		return !anchor
+	})
+
+	// tier 1's build comes in, as the build machine's outcome is taken in.
+	b := plan().Modules["b"]
+	if b == nil || b.Build == "" {
+		t.Fatalf("tier 1 was not asked: %+v", 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 both machines", func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(sends) == 2
+	})
+	reports("anchor")
+
+	// laptop never reports: its key expires and the new controller names it silent, which ends the rollout.
+	eventually(t, "the rollout ended, naming laptop silent and not anchor", func() bool {
+		p := plan()
+		return p.State == inventory.PlanDone
+	})
+	p = plan()
+	if !strings.Contains(p.Note, "laptop") || strings.Contains(p.Note, "anchor") || !strings.Contains(p.Note, "said nothing") {
+		t.Errorf("the plan ended with the note %q; want laptop named silent and not anchor", p.Note)
+	}
+	eventually(t, "no key left on the bus", func() bool { return len(heldKeys()) == 0 })
+
+	// No tier asked twice and none skipped: one ask for tier 1 (tier 0 was built when the plan was kept), each tier sent
+	// once to both machines, and said as sent once for each.
+	mu.Lock()
+	if len(asks) != 1 || !strings.HasPrefix(asks[0], "novox/b#") {
+		t.Errorf("asks = %v; want tier 1's module asked once", asks)
+	}
+	if len(sends) != 2 || strings.Join(sends[0], ",") != "anchor,laptop" || strings.Join(sends[1], ",") != "anchor,laptop" {
+		t.Errorf("sends = %v; want both machines sent each tier once", sends)
+	}
+	mu.Unlock()
+	said := map[string]int{}
+	for deadline := time.After(5 * time.Second); len(said) < 4; {
+		select {
+		case m := <-sentHeard:
+			var body planSentBody
+			if err := json.Unmarshal(m.Data, &body); err != nil {
+				t.Fatal(err)
+			}
+			said[fmt.Sprintf("%d/%s", body.Tier, body.Node)]++
+		case <-deadline:
+			t.Fatalf("sent was said for %v; want tiers 0 and 1 to both machines", said)
+		}
+	}
+	time.Sleep(300 * time.Millisecond)
+	for len(sentHeard) > 0 {
+		var body planSentBody
+		_ = json.Unmarshal((<-sentHeard).Data, &body)
+		said[fmt.Sprintf("%d/%s", body.Tier, body.Node)]++
+	}
+	for k, n := range said {
+		if n != 1 {
+			t.Errorf("sent for %s was said %d times", k, n)
+		}
+	}
+	// And every move of the plan was said as plan-moved.
+	saved.noUnsaid(t, open)
+}