cmd/mesh-controller/release_plan.go

@@ -250,6 +250,68 @@ func appliedEach(module string, since func(node string) time.Time, running []str
 	return len(waiting) == 0, waiting
 }
 
+// gateNeed is a module the next tier waits for: the machines running it, and from when each must have reported.
+type gateNeed struct {
+	module  string
+	running []string
+	since   func(node string) time.Time
+}
+
+// silentNodes is the machines a walk sent and heard nothing from past ApplySilentAfter, on whichever tier it sent
+// them (novox/hq ADR 0360 §5, issue 1160): the one rule both the gates between tiers and the walk's end read, through
+// walkReports. A machine that reports on its send is not silent from then on.
+func silentNodes(p inventory.Plan, reports []inventory.Reported, now time.Time, prov inventory.ProvLookup) map[string]bool {
+	_, _, silent := walkReports(p, reports, now, prov)
+	out := map[string]bool{}
+	for _, n := range silent {
+		out[n] = true
+	}
+	return out
+}
+
+// withoutSilent is the machines of running that are not silent: a later tier is not sent to a machine the walk heard
+// nothing from, which keeps its want of the plan's builds for the send it gets when it is heard again (ADR 0360 §5).
+func withoutSilent(running []string, silent map[string]bool) []string {
+	return slices.DeleteFunc(slices.Clone(running), func(n string) bool { return silent[n] })
+}
+
+// settleTierGate is the wait between tiers on the reports of the machines running what the next tier needs: what is
+// still waited for ("module on a, b"), the silent machines left out of the wait, sorted, and whether every machine the
+// gate waits on is silent, so that nothing it needs was delivered anywhere (issue 1160).
+func settleTierGate(needs []gateNeed, reports []inventory.Reported, silent map[string]bool) (waiting, quiet []string, lost bool) {
+	all, silentAll := 0, 0
+	seenQuiet := map[string]bool{}
+	seenNode := map[string]bool{}
+	for _, need := range needs {
+		for _, n := range need.running {
+			if !seenNode[n] {
+				seenNode[n] = true
+				all++
+				if silent[n] {
+					silentAll++
+				}
+			}
+		}
+		_, on := appliedEach(need.module, need.since, need.running, reports)
+		var still []string
+		for _, n := range on {
+			if silent[n] {
+				if !seenQuiet[n] {
+					seenQuiet[n] = true
+					quiet = append(quiet, n)
+				}
+				continue
+			}
+			still = append(still, n)
+		}
+		if len(still) > 0 {
+			waiting = append(waiting, fmt.Sprintf("%s on %s", need.module, strings.Join(still, ", ")))
+		}
+	}
+	sort.Strings(quiet)
+	return waiting, quiet, all > 0 && silentAll == all
+}
+
 func applied(module string, builtAt time.Time, running []string, reports []inventory.Reported) (bool, []string) {
 	at := map[string]*time.Time{}
 	for _, r := range reports {
@@ -869,6 +931,13 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 	// being behind. Now every module of the tier whose first machine is the same is sent there in one
 	// send, judged there by one gate — kept on the first of them, the others gated by it — and the rest
 	// of the tier's machines are sent once, together, when the gates have passed.
+	// **A machine silent past the bound is left as it was, on every tier** (novox/hq ADR 0360 §5, issue 1160): the
+	// walk's end names it, and the gates between tiers and the sends of the tiers after do not wait for it or go to it.
+	allReports, err := inv.LastReports(ctx)
+	if err != nil {
+		return false, err
+	}
+	silent := silentNodes(*p, allReports, time.Now().UTC(), provLookup(ctx))
 	var pending []string
 	firstOn := map[string][]string{}
 	runningOf := map[string][]string{}
@@ -903,6 +972,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 		if goes != nil {
 			running = slices.DeleteFunc(running, func(n string) bool { return !goes[n] })
 		}
+		if state.FirstAt == nil {
+			running = withoutSilent(running, silent)
+		}
 		runningOf[m] = running
 		// **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source
 		// of the build every machine running it was sent, it is registered with that build's artifacts —
@@ -1096,13 +1168,14 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 		return changed, nil
 	}
 	// And wait for what the next tier needs running.
+	leftAsWas := ""
 	needed := gates(*p, edges, rollsOut)
 	if len(needed) > 0 {
 		reports, err := inv.LastReports(ctx)
 		if err != nil {
 			return false, err
 		}
-		var waiting []string
+		var needs []gateNeed
 		for _, m := range needed {
 			running, err := inv.Running(ctx, m)
 			if err != nil {
@@ -1133,14 +1206,12 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 				for _, r := range reports {
 					byNode[r.Node] = r
 				}
-				if ok, on := appliedEach(m, func(node string) time.Time {
+				needs = append(needs, gateNeed{m, running, func(node string) time.Time {
 					if s := byNode[node].Sent; s != nil {
 						return *s
 					}
 					return time.Time{}
-				}, running, reports); !ok {
-					waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", ")))
-				}
+				}})
 				continue
 			}
 			// **Each machine from its own send** (novox/hq issue 256). With one machine first (ADR 0218)
@@ -1160,9 +1231,23 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 				}
 				return at
 			}
-			if ok, on := appliedEach(m, sinceFor, running, reports); !ok {
-				waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", ")))
-			}
+			needs = append(needs, gateNeed{m, running, sinceFor})
+		}
+		waiting, quiet, lost := settleTierGate(needs, reports, silent)
+		if lost {
+			// Every machine the next tier waits on is silent: nothing it needs was delivered anywhere, so the plan does
+			// not go on as if it had been (issue 1160).
+			p.State = inventory.PlanFailed
+			p.Note = fmt.Sprintf("tier %d built; every machine it waits on said nothing on its send within %s: %s",
+				p.Tier, applySilentAfter, strings.Join(quiet, ", "))
+			fmt.Printf("%s: %s\n", p.ID, p.Note)
+			return true, nil
+		}
+		if len(quiet) > 0 {
+			fmt.Printf("%s: tier %d: %s said nothing on its send within %s and is left as it was; the plan goes on without it\n",
+				p.ID, p.Tier, strings.Join(quiet, ", "), applySilentAfter)
+			leftAsWas = fmt.Sprintf("%s said nothing on its send within %s and is left as it was", strings.Join(quiet, ", "),
+				applySilentAfter)
 		}
 		if len(waiting) > 0 {
 			note := "tier " + fmt.Sprint(p.Tier) + " built; waiting for " + strings.Join(waiting, "; ") + " to be applied"
@@ -1204,6 +1289,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
 			carryUserList(ctx, open, p)
 		}
 	}
+	if end == "" {
+		end = leftAsWas
+	}
 	p.Tier++
 	p.State = inventory.PlanBuilding
 	p.Note = end

cmd/mesh-controller/silent_mid_tier_integration_test.go

@@ -0,0 +1,202 @@
+//go:build integration
+
+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"
+)
+
+// 1160/AC5, live, with the store and a real bus and no ticker: a plan of two tiers whose first tier has a machine that
+// never reports, and whose second tier waits on the first (the plan's own order), passes its first tier after the bound,
+// the silent key's expiry the only trigger, and sends its second tier to the machine that reported alone (issue 1160).
+func TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers(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() })
+
+	wasSaved := inventory.PlanSaved
+	inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open)
+	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)
+		}
+	}
+	plan := func() inventory.Plan {
+		t.Helper()
+		p, err := inv.PlanByID(ctx, "plan-1160")
+		if err != nil {
+			t.Fatal(err)
+		}
+		return p
+	}
+
+	built := time.Now().UTC()
+	p := inventory.Plan{ID: "plan-1160", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: time.Now().UTC(),
+		State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
+		Delivery: &inventory.PlanDelivery{Order: []inventory.Edge{{From: "b", To: "a", Kind: edgeChangeOrder}}},
+		Modules: map[string]*inventory.PlanModule{
+			"a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"},
+			"b": {},
+		}}
+	first, stopFirst := context.WithCancel(ctx)
+	defer stopFirst()
+	startRolloutEvents(first, open, js)
+	if err := inv.SavePlan(ctx, &p); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "tier 0 sent to both machines", func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(sends) == 1
+	})
+	reports("anchor")
+	time.Sleep(time.Second)
+	if got := plan().Tier; got != 0 {
+		t.Fatalf("1160/AC5: the plan went on to tier %d before the bound, with laptop not heard from", got)
+	}
+	eventually(t, "tier 1 asked once laptop's key expired", func() bool { return plan().Tier == 1 })
+	if note := plan().Note; !strings.Contains(note, "laptop") || strings.Contains(note, "anchor") {
+		t.Errorf("1160/AC5: the plan says %q; want laptop named silent and not anchor", note)
+	}
+	b := plan().Modules["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 anchor alone", func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(sends) == 2 && strings.Join(sends[1], ",") == "anchor"
+	})
+}

cmd/mesh-controller/silent_mid_tier_test.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/inventory"
+)
+
+// A machine silent past the bound at a tier between the first and the last is named silent and left as it was, as the
+// last tier's is (novox/hq ADR 0282 §1, ADR 0360 §4-5, issue 1160): one rule, walkReports, read by the walk's end and by
+// the gates between tiers.
+
+func silentWalk(now time.Time) (inventory.Plan, []inventory.Reported) {
+	sent := now.Add(-time.Minute)
+	long := now.Add(-inventory.ApplySilentAfter - time.Minute)
+	at := now.Add(-10 * time.Second)
+	p := inventory.Plan{Tier: 0, Tiers: [][]string{{"engine"}, {"app"}}, Modules: map[string]*inventory.PlanModule{
+		"engine": {Rest: map[string]inventory.SentDeclaration{
+			"anchor": {Digest: "d-anchor", At: sent},
+			"g14":    {Digest: "d-g14", At: long},
+			"fresh":  {Digest: "d-fresh", At: sent}}},
+		"app": {}}}
+	reports := []inventory.Reported{{Node: "anchor", Outcome: inventory.OutcomeApplied, At: &at, Declared: "d-anchor"}}
+	return p, reports
+}
+
+func needsOf(running ...string) []gateNeed {
+	return []gateNeed{{module: "engine", running: running, since: func(string) time.Time { return time.Time{} }}}
+}
+
+// 1160/AC1: at a middle tier the machine silent past the bound is named, not waited for, and the plan goes on with the
+// machines that reported; one still inside the bound is waited for.
+func TestAMiddleTierGateNamesTheSilentMachineAndGoesOnWithThoseThatReported(t *testing.T) {
+	now := time.Now().UTC()
+	p, reports := silentWalk(now)
+	silent := silentNodes(p, reports, now, nil)
+	waiting, quiet, lost := settleTierGate(needsOf("anchor", "g14"), reports, silent)
+	if len(waiting) != 0 || !slices.Equal(quiet, []string{"g14"}) || lost {
+		t.Fatalf("1160/AC1: waiting %v, silent %v, lost %v; want g14 named silent and nothing waited for", waiting, quiet, lost)
+	}
+	waiting, quiet, _ = settleTierGate(needsOf("anchor", "g14", "fresh"), reports, silent)
+	if len(waiting) != 1 || waiting[0] != "engine on fresh" || !slices.Equal(quiet, []string{"g14"}) {
+		t.Fatalf("1160/AC1: a machine inside the bound is still waited for: waiting %v, silent %v", waiting, quiet)
+	}
+}
+
+// 1160/AC2: the walk's end and the gate read the same silent machines.
+func TestTheGateAndTheWalksEndShareOneRuleOfSilence(t *testing.T) {
+	now := time.Now().UTC()
+	p, reports := silentWalk(now)
+	_, _, ended := walkReports(p, reports, now, nil)
+	got := silentNodes(p, reports, now, nil)
+	if len(got) != len(ended) || !got["g14"] || !slices.Equal(ended, []string{"g14"}) {
+		t.Fatalf("1160/AC2: the gate reads %v silent, the walk's end %v", got, ended)
+	}
+}
+
+// 1160/AC3: a later tier is not sent to a machine silent.
+func TestALaterTierIsNotSentToASilentMachine(t *testing.T) {
+	got := withoutSilent([]string{"anchor", "g14"}, map[string]bool{"g14": true})
+	if !slices.Equal(got, []string{"anchor"}) {
+		t.Fatalf("1160/AC3: a later tier goes to %v; want anchor alone", got)
+	}
+	in := []string{"anchor", "g14"}
+	withoutSilent(in, map[string]bool{"g14": true})
+	if !slices.Equal(in, []string{"anchor", "g14"}) {
+		t.Fatalf("1160/AC3: the machines running it were changed: %v", in)
+	}
+}
+
+// 1160/AC4: every machine the gate waits on silent: the plan is lost, not gone on as if delivered.
+func TestAGateWhoseEveryMachineIsSilentIsLost(t *testing.T) {
+	now := time.Now().UTC()
+	p, reports := silentWalk(now)
+	reports = nil
+	p.Modules["engine"].Rest = map[string]inventory.SentDeclaration{
+		"g14": {Digest: "d", At: now.Add(-time.Hour)}, "laptop": {Digest: "d", At: now.Add(-time.Hour)}}
+	silent := silentNodes(p, reports, now, nil)
+	_, quiet, lost := settleTierGate(needsOf("g14", "laptop"), reports, silent)
+	if !lost || !slices.Equal(quiet, []string{"g14", "laptop"}) {
+		t.Fatalf("1160/AC4: lost %v, silent %v; want both named and the gate lost", lost, quiet)
+	}
+}

testdata/store-tests

@@ -163,3 +163,4 @@
 ./internal/inventory TestASendEmptiesTheColumnSoALaterCopyLeavesAClearedKeyBare
 ./internal/inventory TestWhetherAMachineReadsModulesIsItsLatestWord
 ./cmd/mesh-controller TestAComposedDeclarationCarriesModulesAndBuildsOnlyForANodeThatReadsThem
+./cmd/mesh-controller TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers