issue 974: review 1: a verb's end, a provisioner client's word and a failed key read each ask or retry a look; the silent bucket ages; the acceptance test starts the rollout through the serving wiring

cmd/mesh-controller/delivery.go

@@ -977,7 +977,7 @@ func deliveryCommand(ctx context.Context, args []string) error {
 		}); err != nil {
 			return err
 		}
-		fmt.Printf("%s (%s at %s) is let go by %s; its first tier is asked at the next pass\n", p.ID, p.Repository,
+		fmt.Printf("%s (%s at %s) is let go by %s; its first tier is asked when the controller next looks at the plans, which it does as this ends\n", p.ID, p.Repository,
 			short(p.Commit), *by)
 		return nil
 	case "stop":

cmd/mesh-controller/plan_looks.go

@@ -160,6 +160,33 @@ func startPlanLooks(ctx context.Context, open *stores) {
 // ApplySilentAfter, a variable so a test of the wait needs no quarter of an hour.
 var applySilentAfter = inventory.ApplySilentAfter
 
+// startRolloutEvents starts what a rollout advances on, for as long as the context lives (novox/hq ADR 0406 §5, issue 974):
+// the wait for each machine's report as a key on the bus that expires (where the bus will not take it a timer of its own
+// for that wait looks instead), a provisioner client's words, and the look at the plans, asked for first so a plan left
+// by a replaced controller resumes. The serving controller's start and the tests of a rollout call this one.
+func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream) {
+	if err := startRolloutSilent(ctx, js); err != nil {
+		fmt.Printf("plans: the wait for the machines' reports could not be put on the bus (%v); each is looked at by a timer of its own\n", err)
+	}
+	if err := startProvAppliedLooks(ctx, js); err != nil {
+		fmt.Printf("plans: the provisioner clients' words could not be watched (%v); a walk waits for them until the next look\n", err)
+	}
+	startPlanLooks(ctx, open)
+}
+
+// servingPlanSaved is what the serving controller does for each plan it keeps: said as plan-moved, looked at (a build, a
+// send kept or a command's move is followed by the look that asks the next tier, in place of the timer that did), and,
+// for a walk that ended, its merges' check builds forgotten (novox/hq ADR 0350 §5).
+func servingPlanSaved(ctx context.Context, bus link.Bus, open *stores) func(*inventory.Inventory, inventory.Plan) {
+	return func(_ *inventory.Inventory, p inventory.Plan) {
+		sayPlanMoved(ctx, bus, open.inventory, p)
+		planKept(p)
+		if p.WalkSaid == inventory.WalkEnded {
+			go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, p)
+		}
+	}
+}
+
 // planLookPasses is the most times one look goes over the plans again because the first pass changed which are open.
 const planLookPasses = 6
 
@@ -193,6 +220,7 @@ func lookAtPlans(ctx context.Context, open *stores) (failed bool) {
 		}
 		if err := syncSilentKeys(ctx, open); err != nil {
 			fmt.Printf("plans: the wait for the machines' reports could not be kept on the bus: %v\n", err)
+			failed = true // so the look is retried
 		}
 	}()
 	keepWalkPhases(ctx, open)
@@ -365,15 +393,25 @@ func syncSilentKeys(ctx context.Context, open *stores) error {
 			want[w] = at
 		}
 	}
-	if c == nil {
+	return keepSilentKeys(ctx, c, want, now)
-		// No bucket to put them in: each wait is looked at by a timer of its own, once.
+}
+
+// keepSilentKeys puts the keys on the bus as the waits are. Where the bus will not take a key — no bucket, the keys
+// cannot be read, one cannot be put — the wait is looked at by a timer of its own, once, and the error is given back so
+// the look is retried.
+func keepSilentKeys(ctx context.Context, c silentClock, want map[silentWait]time.Time, now time.Time) error {
+	byTimer := func() {
 		for w, sent := range want {
 			silentByTimer(w, sent, silentTTL(sent, now))
 		}
+	}
+	if c == nil {
+		byTimer()
 		return nil
 	}
 	held, err := c.Held(ctx)
 	if err != nil {
+		byTimer()
 		return err
 	}
 	arm, drop := silentDiff(want, held)
@@ -442,6 +480,33 @@ func startRolloutSilent(ctx context.Context, js *broker.JetStream) error {
 	return nil
 }
 
+// startProvAppliedLooks has the plans looked at when a machine's provisioner client says a word (`prov.applied.<node>.<module>`):
+// a send that handed modules to the provisioner is applied only once the client said so too, and without this the walk
+// would wait for its silent key (ADR 0362, ADR 0406 §5, issue 974). New words only: the first look reads the rest.
+func startProvAppliedLooks(ctx context.Context, js *broker.JetStream) error {
+	api, err := jetstream.New(js.Conn())
+	if err != nil {
+		return err
+	}
+	stream, err := api.Stream(ctx, broker.ProvAppliedStream)
+	if err != nil {
+		return fmt.Errorf("the provisioner clients' words: %w", err)
+	}
+	consumer, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{DeliverPolicy: jetstream.DeliverNewPolicy})
+	if err != nil {
+		return fmt.Errorf("watching the provisioner clients' words: %w", err)
+	}
+	consuming, err := consumer.Consume(func(jetstream.Msg) { askToLook() })
+	if err != nil {
+		return fmt.Errorf("watching the provisioner clients' words: %w", err)
+	}
+	go func() {
+		<-ctx.Done()
+		consuming.Stop()
+	}()
+	return nil
+}
+
 // --- the end of a walk whose end is not known ---
 
 var walkEnds = struct {

cmd/mesh-controller/plan_looks_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -257,3 +258,60 @@ func TestAnExpiredSilentKeyAsksForALookAndAReportsDeleteDoesNot(t *testing.T) {
 	}
 	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")
+	}
+}

cmd/mesh-controller/push.go

@@ -183,12 +183,7 @@ func serve(ctx context.Context) (err error) {
 	// Open plans move on events (novox/hq ADR 0406 §5, issue 974): a tier is asked on the `built` and the send of the
 	// tier before it, a walk ends on its machines' reports, a silent machine is a key that expired, and a plan left
 	// by a replaced controller resumes from the first look, which this asks for. No timer looks every so often.
-	// The wait for each machine's report is a key on the bus that expires; where the bus will not take it a timer of its
+	startRolloutEvents(ctx, open, server.JetStream())
-	// own for that wait looks instead.
-	if err := startRolloutSilent(ctx, server.JetStream()); err != nil {
-		fmt.Printf("plans: the wait for the machines' reports could not be put on the bus (%v); each is looked at by a timer of its own\n", err)
-	}
-	startPlanLooks(ctx, open)
 	// And the merge window (novox/hq ADR 0276, ADR 0407): a batch is cut into its walk when the window closes,
 	// which no merge or outcome says. The window is two keys on the bus that expire and a watcher that hears it;
 	// nothing polls (issue 973).
@@ -258,16 +253,7 @@ func serve(ctx context.Context) (err error) {
 	// (novox/hq to-be 45 §9).
 	checkEvents = bus
 	// And every walk this controller keeps in a new state, for the delivery it walks (novox/hq ADR 0239).
-	inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) {
+	inventory.PlanSaved = servingPlanSaved(ctx, bus, open)
-		sayPlanMoved(ctx, bus, open.inventory, p)
-		// And the plan's next step is looked at (ADR 0406 §5): a build, a send kept or a command's move is followed
-		// by the look that asks the next tier, in place of the timer that did.
-		planKept(p)
-		// A walk ended has taken what it reuses: its merges' check builds go (novox/hq ADR 0350 §5).
-		if p.WalkSaid == inventory.WalkEnded {
-			go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, p)
-		}
-	}
 	if err := server.Checks(following{open: open}); err != nil {
 		return err
 	}

cmd/mesh-controller/release_plan.go

@@ -2016,7 +2016,7 @@ func plansCommand(ctx context.Context, args []string) error {
 		}); err != nil {
 			return err
 		}
-		fmt.Printf("%s (%s at %s) is let go by hand; its first tier is asked at the next pass\n", p.ID, p.Repository,
+		fmt.Printf("%s (%s at %s) is let go by hand; its first tier is asked when the controller next looks at the plans, which it does as this ends\n", p.ID, p.Repository,
 			short(p.Commit))
 		return nil
 	}

cmd/mesh-controller/rollout_events_test.go

@@ -56,10 +56,11 @@ func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
 	}
 	t.Cleanup(func() { _ = sub.Unsubscribe() })
 
-	// Every save of a plan is said and asks the look, as the serving controller's hook does.
+	// Every save of a plan is kept, said and looked at by the serving controller's own hook.
 	saved := &movesSaid{revs: map[string][]int64{}}
 	wasSaved := inventory.PlanSaved
-	inventory.PlanSaved = func(i *inventory.Inventory, p inventory.Plan) { saved.hear(i, p); planKept(p) }
+	serving := servingPlanSaved(ctx, checkEvents, open)
+	inventory.PlanSaved = func(i *inventory.Inventory, p inventory.Plan) { saved.hear(i, p); serving(i, 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.
@@ -176,10 +177,7 @@ func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
 	// 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 {
+	startRolloutEvents(first, open, js)
-		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"}},
@@ -204,10 +202,7 @@ func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
 
 	second, stopSecond := context.WithCancel(ctx)
 	defer stopSecond()
-	if err := startRolloutSilent(second, js); err != nil {
+	startRolloutEvents(second, open, js)
-		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()
@@ -279,3 +274,72 @@ func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
 	// And every move of the plan was said as plan-moved.
 	saved.noUnsaid(t, open)
 }
+
+// Acceptance of the verb path (issue 974, review 1), live, with the store and a bus: a walk the delivery's owner lets
+// go by the verb `deliver` after the operator's approval is asked its first tier with no other event. The verb runs as a
+// child process, whose saves the serving controller does not hear; the handler asks the look as the verb ends.
+func TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent(t *testing.T) {
+	open := aMesh(t)
+	ctx := t.Context()
+	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 })
+	asked := asksRecorded(t)
+	register(t, open, catalogue.Manifest{Module: "app", Version: "1"})
+	if _, err := assign(ctx, open, "anchor", "app"); err != nil {
+		t.Fatal(err)
+	}
+	cut := time.Now().UTC()
+	waiting := inventory.Plan{ID: "plan-974-deliver", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c8c8c8c8",
+		Created: cut, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
+		Modules:  map[string]*inventory.PlanModule{"app": {}},
+		Delivery: &inventory.PlanDelivery{Awaits: catalogue.DeliverySeat},
+		Times:    &inventory.PlanTimes{Cut: &cut, Class: inventory.ClassLeaf}}
+	if err := open.inventory.SavePlan(ctx, &waiting); err != nil {
+		t.Fatal(err)
+	}
+
+	// The serving controller, as serve starts it.
+	wasSaved := inventory.PlanSaved
+	inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open)
+	t.Cleanup(func() { inventory.PlanSaved = wasSaved })
+	started, stop := context.WithCancel(ctx)
+	defer stop()
+	startRolloutEvents(started, open, js)
+	time.Sleep(500 * time.Millisecond) // the first look: the walk waits, and is asked nothing
+
+	// The verb as its child process runs it: the delivery command, whose saves are not the serving controller's.
+	wasRun := runTheVerb
+	runTheVerb = func(ctx context.Context, argv []string) (verbAnswer, error) {
+		was := inventory.PlanSaved
+		inventory.PlanSaved = func(*inventory.Inventory, inventory.Plan) {}
+		defer func() { inventory.PlanSaved = was }()
+		return verbAnswer{}, deliveryCommand(ctx, []string{"go", waiting.ID, "--why", "its turn"})
+	}
+	t.Cleanup(func() { runTheVerb = wasRun })
+	handlers, _, err := seatToolHandlers()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := handlers["deliver"](ctx, []byte(`{"plan":"`+waiting.ID+`","why":"its turn"}`)); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "the walk's first tier asked, with no event but the verb's end", func() bool {
+		p, err := open.inventory.PlanByID(ctx, waiting.ID)
+		return err == nil && p.Modules["app"] != nil && p.Modules["app"].Build != "" && len(*asked) == 1
+	})
+}

cmd/mesh-controller/seatverbs.go

@@ -1398,12 +1398,24 @@ func seatToolHandlers() (map[string]link.ToolHandler, []string, error) {
 				// reloading its user list forgets the answer it was about to permit (novox/hq issue 265).
 				link.Acknowledge(ctx)
 			}
-			return runVerb(ctx, argv)
+			answer, err := runTheVerb(ctx, argv)
+			// **A plan the verb moved is looked at as it ends** (novox/hq ADR 0406 §5, issue 974): the verb ran as a
+			// child process, whose saves this process does not hear, so `deliver` after an approval, a stop, a retry
+			// and a send from a verb would otherwise wait for an unrelated event. Even when it failed: it may have
+			// kept a plan before it did.
+			if !readingVerbs[verb] && !(verb == "plans" && !actsOnAPlan(args)) {
+				askToLook()
+			}
+			return answer, err
 		}
 	}
 	return handlers, behind, nil
 }
 
+// runTheVerb is how the serving controller runs a verb: as a command of its own. A variable so a test of what follows a
+// verb needs no child process.
+var runTheVerb = runVerb
+
 // actsOnAPlan is `plans` asked to stop, close or retry one rather than to show them.
 func actsOnAPlan(args map[string]any) bool {
 	for _, act := range []string{"stop", "close", "retry"} {

cmd/mesh-controller/verb_looks_test.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// A plan moved by a seat verb is looked at as the verb ends (issue 974, review 1): the verb runs as a child process
+// whose saves the serving controller does not hear, so `deliver`, which lets a walk go after the operator's approval,
+// would leave its first tier waiting for an unrelated event.
+func TestAVerbThatMayMoveAPlanHasTheServingControllerLookWhenItEnds(t *testing.T) {
+	var looks, failures 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)
+
+	was := runTheVerb
+	runTheVerb = func(context.Context, []string) (verbAnswer, error) { return verbAnswer{}, nil }
+	t.Cleanup(func() { runTheVerb = was })
+	handlers, _, err := seatToolHandlers()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, c := range []struct {
+		verb string
+		args string
+		look bool
+	}{
+		{"deliver", `{"plan":"p"}`, true},
+		{"plans", `{"stop":"plan-1","why":"w"}`, true},
+		{"plans", `{}`, false},
+		{"nodes", `{}`, false},
+	} {
+		before := looks.Load()
+		h := handlers[c.verb]
+		if h == nil {
+			t.Fatalf("no verb %s", c.verb)
+		}
+		if _, err := h(ctx, []byte(c.args)); err != nil {
+			t.Fatalf("%s %s: %v", c.verb, c.args, err)
+		}
+		time.Sleep(100 * time.Millisecond)
+		if got := looks.Load() > before; got != c.look {
+			t.Errorf("%s %s: look asked = %v; want %v", c.verb, c.args, got, c.look)
+		}
+	}
+}

internal/broker/controller_buckets.go

@@ -97,6 +97,11 @@ const MergeWindowMarkerTTL = time.Minute
 // RolloutSilentMarkerTTL is how long the marker an expired silent key leaves is kept, for the same reason.
 const RolloutSilentMarkerTTL = time.Minute
 
+// RolloutSilentKeptFor is the age of the silent keys' bucket: every report deletes a key and the delete leaves a marker, which
+// with one value per key would stay for ever and fill the bucket. A key lives a quarter of an hour at most, so an hour
+// takes nothing live.
+const RolloutSilentKeptFor = time.Hour
+
 // AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own.
 const AskedKeptFor = 30 * 24 * time.Hour
 
@@ -251,6 +256,7 @@ func (j *JetStream) EnsureControllerBuckets() error {
 			"lives the time a machine may stay silent and is deleted by its report; one that expires names the " +
 			"machine silent; written by the controller alone",
 		History:        1,
+		TTL:            RolloutSilentKeptFor,
 		MaxValueSize:   4 << 10,
 		MaxBytes:       8 << 20,
 		Storage:        jetstream.FileStorage,

internal/broker/controller_buckets_test.go

@@ -389,6 +389,42 @@ func TestTheRolloutsSilentKeysExpireAsAPurgeAndAReportsDeleteIsNot(t *testing.T)
 	}
 }
 
+// **The silent keys' bucket ages**, or its delete markers would fill it (issue 974, review 1): every report deletes a key,
+// the delete leaves a marker, and with one value per key and no age the markers stay for ever. A bucket made before the
+// age was set is brought to it by the next start, which the server accepts as an update of an existing bucket.
+func TestTheSilentKeysBucketAgesAndAnOlderOneIsBroughtToIt(t *testing.T) {
+	js, err := Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	api, err := jetstream.New(js.Conn())
+	if err != nil {
+		t.Fatal(err)
+	}
+	// As the first build made it: no age.
+	if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: RolloutSilentBucket, History: 1,
+		MaxValueSize: 4 << 10, MaxBytes: 8 << 20, Storage: jetstream.FileStorage, LimitMarkerTTL: RolloutSilentMarkerTTL}); err != nil {
+		t.Fatal(err)
+	}
+	if err := js.EnsureControllerBuckets(); err != nil {
+		t.Fatalf("a bucket made without the age is not brought to it: %v", err)
+	}
+	kv, err := api.KeyValue(ctx, RolloutSilentBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	status, err := kv.Status(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status.TTL() != RolloutSilentKeptFor {
+		t.Errorf("the bucket ages at %v; want %v", status.TTL(), RolloutSilentKeptFor)
+	}
+}
+
 // **The merges have a consumer of their own, which a reset of the events consumer leaves alone** (issue 973, novox/hq
 // ADR 0407 decision 3): healer H4 re-makes the events consumer from now, dropping what it holds pending; a merge
 // waiting in the merges consumer is not that consumer's, and nothing would catch it up.

testdata/store-tests

@@ -135,3 +135,4 @@
 ./cmd/mesh-controller TestABatchWindowClosesByTheExpiryOfItsKey
 ./cmd/mesh-controller TestEveryMoveAMergeAndTheCutMakeOfAPlanIsSaid
 ./cmd/mesh-controller TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt
+./cmd/mesh-controller TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent