diff --git a/cmd/mesh-controller/delivery.go b/cmd/mesh-controller/delivery.go index 17ba3c4a..42b0ee7b 100644 --- a/cmd/mesh-controller/delivery.go +++ b/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": diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index 6ae5db79..6e4f155d 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/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 { - // No bucket to put them in: each wait is looked at by a timer of its own, once. + return keepSilentKeys(ctx, c, want, now) +} + +// 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..`): +// 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 { diff --git a/cmd/mesh-controller/plan_looks_test.go b/cmd/mesh-controller/plan_looks_test.go index e6612960..3f4544eb 100644 --- a/cmd/mesh-controller/plan_looks_test.go +++ b/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") + } +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index af6d40e1..66099d0a 100644 --- a/cmd/mesh-controller/push.go +++ b/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 - // 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) + startRolloutEvents(ctx, open, server.JetStream()) // 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) { - 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) - } - } + inventory.PlanSaved = servingPlanSaved(ctx, bus, open) if err := server.Checks(following{open: open}); err != nil { return err } diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index c0b90e63..e10eb7cf 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/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 } diff --git a/cmd/mesh-controller/rollout_events_test.go b/cmd/mesh-controller/rollout_events_test.go index 7617a7dd..f3809bba 100644 --- a/cmd/mesh-controller/rollout_events_test.go +++ b/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 { - t.Fatal(err) - } - startPlanLooks(first, open) + startRolloutEvents(first, open, js) 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 { - t.Fatal(err) - } - startPlanLooks(second, open) + startRolloutEvents(second, open, js) // 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 + }) +} diff --git a/cmd/mesh-controller/seatverbs.go b/cmd/mesh-controller/seatverbs.go index 101db41c..4a2e0552 100644 --- a/cmd/mesh-controller/seatverbs.go +++ b/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"} { diff --git a/cmd/mesh-controller/verb_looks_test.go b/cmd/mesh-controller/verb_looks_test.go new file mode 100644 index 00000000..d16e190a --- /dev/null +++ b/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) + } + } +} diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index b47d9111..7d8ba4e4 100644 --- a/internal/broker/controller_buckets.go +++ b/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, diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index e34e0d3e..4b337dc1 100644 --- a/internal/broker/controller_buckets_test.go +++ b/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. diff --git a/testdata/store-tests b/testdata/store-tests index 8a116cec..81cc53c1 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -135,3 +135,4 @@ ./cmd/mesh-controller TestABatchWindowClosesByTheExpiryOfItsKey ./cmd/mesh-controller TestEveryMoveAMergeAndTheCutMakeOfAPlanIsSaid ./cmd/mesh-controller TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt +./cmd/mesh-controller TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent