From fd522f432200d6b72d5c9a7ecc09f84e64973e7b Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 13:18:27 +0200 Subject: [PATCH] issue 974: a rollout's wait for a machine is a key on the bus that its report deletes and its expiry names silent --- cmd/mesh-controller/rollout_silent_bus.go | 121 ++++++++++++++++++ .../rollout_silent_bus_test.go | 120 +++++++++++++++++ internal/broker/controller_buckets.go | 28 +++- internal/broker/controller_buckets_test.go | 67 ++++++++++ internal/broker/nats.go | 5 + internal/broker/testdata/composed.conf | 2 +- internal/broker/writers.go | 10 +- 7 files changed, 346 insertions(+), 7 deletions(-) create mode 100644 cmd/mesh-controller/rollout_silent_bus.go create mode 100644 cmd/mesh-controller/rollout_silent_bus_test.go diff --git a/cmd/mesh-controller/rollout_silent_bus.go b/cmd/mesh-controller/rollout_silent_bus.go new file mode 100644 index 00000000..ca21ce66 --- /dev/null +++ b/cmd/mesh-controller/rollout_silent_bus.go @@ -0,0 +1,121 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A rollout's wait for a machine is a key on the bus (novox/hq ADR 0406 §5, issue 974): each send to a machine puts +// a key `..` that lives as long as the machine may stay silent (ApplySilentAfter); the machine's +// report on that send deletes it, so only a missing report lets it expire; and the expiry, heard by a watcher, +// ends the wait for that machine, which the plan then names silent. No timer looks every so often whether a +// machine has reported. + +// silentWait is one machine's wait on one send of a plan's tier. +type silentWait struct { + Plan string + Tier int + Node string +} + +// silentKey is a wait's key. A plan id may hold a dot, so it is read from the right (parseSilentKey). +func silentKey(w silentWait) string { return w.Plan + "." + strconv.Itoa(w.Tier) + "." + w.Node } + +// parseSilentKey reads a key back: the node is the last token, the tier the one before, the plan what is left. +func parseSilentKey(key string) (silentWait, bool) { + last := strings.LastIndexByte(key, '.') + if last < 1 || last == len(key)-1 { + return silentWait{}, false + } + rest, node := key[:last], key[last+1:] + mid := strings.LastIndexByte(rest, '.') + if mid < 1 || mid == len(rest)-1 { + return silentWait{}, false + } + tier, err := strconv.Atoi(rest[mid+1:]) + if err != nil || tier < 0 { + return silentWait{}, false + } + return silentWait{Plan: rest[:mid], Tier: tier, Node: node}, true +} + +// silentClock is the bus's side of the wait for a machine's report. +type silentClock interface { + // Arm puts a wait's key, to live ttl: made again if the machine is sent again. + Arm(ctx context.Context, w silentWait, ttl time.Duration) error + // Reported takes a wait's key away: the machine reported on the send, and nothing is to wake the controller. + Reported(ctx context.Context, w silentWait) error +} + +// busSilent is the clock on the controller's bucket. +type busSilent struct { + kv jetstream.KeyValue +} + +// newBusSilent opens the bucket and starts the watcher, which names each wait whose key expired until the context +// ends. Only expiry wakes it: the marker of an expiry is a purge, and a key a report deleted, or one Arm made again, +// leaves a delete (the vendored client reads the server's MaxAge marker as a purge, kv.go:971). +func newBusSilent(ctx context.Context, api jetstream.JetStream, silent func(silentWait)) (*busSilent, error) { + kv, err := api.KeyValue(ctx, broker.RolloutSilentBucket) + if err != nil { + return nil, fmt.Errorf("the rollout's silent keys' bucket: %w", err) + } + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + return nil, fmt.Errorf("watching the rollout's silent keys: %w", err) + } + go func() { + defer w.Stop() + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil || e.Operation() != jetstream.KeyValuePurge { + continue // a put, or a delete: a report's, or a send made again + } + if wait, ok := parseSilentKey(e.Key()); ok && silent != nil { + silent(wait) + } + } + } + }() + return &busSilent{kv: kv}, nil +} + +func (s *busSilent) Arm(ctx context.Context, w silentWait, ttl time.Duration) error { + key := silentKey(w) + // Made again, not changed: a key's time to live is set when it is made. + if err := s.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + _, err := s.kv.Create(ctx, key, []byte(time.Now().UTC().Format(time.RFC3339)), jetstream.KeyTTL(leastSecond(ttl))) + return err +} + +func (s *busSilent) Reported(ctx context.Context, w silentWait) error { + if err := s.kv.Delete(ctx, silentKey(w)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil +} + +// leastSecond rounds a time to live up to a whole second, at least one: the server rounds a key's time down to +// whole seconds, so a fraction would close the key before the time it stands for. +func leastSecond(d time.Duration) time.Duration { + if d < time.Second { + return time.Second + } + return (d + time.Second - 1) / time.Second * time.Second +} diff --git a/cmd/mesh-controller/rollout_silent_bus_test.go b/cmd/mesh-controller/rollout_silent_bus_test.go new file mode 100644 index 00000000..b9cc46f0 --- /dev/null +++ b/cmd/mesh-controller/rollout_silent_bus_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/testbus" +) + +// silentOnABus is the rollout's silent keys on a real bus, and the waits its watcher heard expire. +func silentOnABus(t *testing.T) (*busSilent, <-chan silentWait) { + t.Helper() + 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) + } + conn, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(conn.Close) + api, err := jetstream.New(conn) + if err != nil { + t.Fatal(err) + } + heard := make(chan silentWait, 16) + ctx, stop := context.WithCancel(context.Background()) + t.Cleanup(stop) + s, err := newBusSilent(ctx, api, func(w silentWait) { heard <- w }) + if err != nil { + t.Fatal(err) + } + return s, heard +} + +func hearsSilent(t *testing.T, heard <-chan silentWait, within time.Duration) (silentWait, bool) { + t.Helper() + select { + case w := <-heard: + return w, true + case <-time.After(within): + return silentWait{}, false + } +} + +// A machine that never reports on its send is named silent when its key expires (issue 974, novox/hq ADR 0406 §5): +// the watcher names the plan, the tier and the machine, which a plan id or a machine name with a dot does not blur. +func TestAKeyLeftAloneExpiresAndTheWatcherNamesTheMachineSilent(t *testing.T) { + s, heard := silentOnABus(t) + want := silentWait{Plan: "plan.7", Tier: 2, Node: "ace"} + if err := s.Arm(t.Context(), want, time.Second); err != nil { + t.Fatal(err) + } + if got, ok := hearsSilent(t, heard, 10*time.Second); !ok || got != want { + t.Fatalf("heard %+v (%v); want %+v", got, ok, want) + } +} + +// A machine that reports in time is never named silent: the report deletes the key, and a delete is not an expiry. +func TestAReportDeletesTheKeyAndNoOneIsNamedSilent(t *testing.T) { + s, heard := silentOnABus(t) + ctx := t.Context() + w := silentWait{Plan: "plan-1", Tier: 0, Node: "g14"} + if err := s.Arm(ctx, w, time.Second); err != nil { + t.Fatal(err) + } + if err := s.Reported(ctx, w); err != nil { + t.Fatal(err) + } + if got, ok := hearsSilent(t, heard, 2500*time.Millisecond); ok { + t.Fatalf("%+v was named silent after it reported", got) + } + // Reporting twice, or on a send that put no key, is no error. + if err := s.Reported(ctx, w); err != nil { + t.Fatalf("a second report: %v", err) + } +} + +// A send made again to a machine puts its key again with the full time, and only the last send's key can expire. +func TestASendMadeAgainPutsTheKeyAgain(t *testing.T) { + s, heard := silentOnABus(t) + ctx := t.Context() + w := silentWait{Plan: "plan-3", Tier: 1, Node: "novox"} + if err := s.Arm(ctx, w, 2*time.Second); err != nil { + t.Fatal(err) + } + time.Sleep(1200 * time.Millisecond) + if err := s.Arm(ctx, w, 2*time.Second); err != nil { + t.Fatal(err) + } + if got, ok := hearsSilent(t, heard, 1500*time.Millisecond); ok { + t.Fatalf("%+v expired at the first send's time, not the second's", got) + } + if got, ok := hearsSilent(t, heard, 5*time.Second); !ok || got != w { + t.Fatalf("heard %+v (%v); want %+v after the second send's time", got, ok, w) + } +} + +func TestASilentKeyIsParsedFromTheRightOfTheNode(t *testing.T) { + for _, w := range []silentWait{{"plan-1", 0, "ace"}, {"a.b", 12, "g14"}} { + got, ok := parseSilentKey(silentKey(w)) + if !ok || got != w { + t.Errorf("%+v became %+v (%v)", w, got, ok) + } + } + for _, bad := range []string{"", "plan", "plan.x.ace", "plan..ace", ".1.ace"} { + if _, ok := parseSilentKey(bad); ok { + t.Errorf("%q was read as a wait", bad) + } + } +} diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 1c534d0d..b47d9111 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -43,6 +43,12 @@ var ( // `.most`, made once when it opens and living the window's most. The first to expire closes the // batch, and the controller's watcher hears the delete marker. Written by the lease holder alone. MergeWindowBucket = BucketName(ControllerSeat, "merge-window") + // RolloutSilentBucket holds the wait of every machine a rollout has sent a tier to (novox/hq ADR 0406 §5, issue + // 974): per send a key `..` that lives ApplySilentAfter, made when the controller sends that + // machine and deleted by the machine's report on that send. Only a missing report lets it expire, and the + // controller's watcher hearing the expiry ends the wait and names the machine silent. Written by the lease + // holder alone. + RolloutSilentBucket = BucketName(ControllerSeat, "rollout-silent") ) // The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of @@ -88,6 +94,9 @@ const ( // that is up to hear it, short enough to leave nothing behind. 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 + // 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 @@ -116,13 +125,13 @@ func IsControllerBucket(bucket string) bool { return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket || bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket || bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || - bucket == MergeWindowBucket + bucket == MergeWindowBucket || bucket == RolloutSilentBucket } // ControllerBuckets are the controller's own buckets, in the order they are asserted. func ControllerBuckets() []string { return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, - AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket} + AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -234,6 +243,21 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", MergeWindowBucket, err) } + // **The rollout's silent keys: keys that expire** (ADR 0406 §5, issue 974). A key lives as long as the machine + // may stay silent; the machine's report deletes it, and a marker of expiry is what the watcher hears. + if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: RolloutSilentBucket, + Description: "the wait of every machine a rollout sent a tier to (novox/hq ADR 0406): `..` " + + "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, + MaxValueSize: 4 << 10, + MaxBytes: 8 << 20, + Storage: jetstream.FileStorage, + LimitMarkerTTL: RolloutSilentMarkerTTL, + }); err != nil { + return fmt.Errorf("asserting bucket %s: %w", RolloutSilentBucket, err) + } // **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose // marker lives a day; a key that expired would be a record the bus forgot while the store still holds it. for _, m := range []struct{ bucket, why string }{ diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 65dcd948..e34e0d3e 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -322,6 +322,73 @@ func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) { } } +// **A rollout's wait for a machine is a bucket whose keys expire** (novox/hq ADR 0406 §5, issue 974): a key made with a +// time to live goes when it is up, and a watcher hears it go as a purge; a key a report deleted is a delete. The +// controller may write it, and it is no undeclared state. +func TestTheRolloutsSilentKeysExpireAsAPurgeAndAReportsDeleteIsNot(t *testing.T) { + controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(controller.Publish, "$KV."+RolloutSilentBucket+".>") { + t.Error("the controller may not write the rollout's silent keys' bucket") + } + if !IsControllerBucket(RolloutSilentBucket) || !slices.Contains(ControllerBuckets(), RolloutSilentBucket) { + t.Error("the rollout's silent keys' bucket is not asserted as the controller's own") + } + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.KeyValue(ctx, RolloutSilentBucket) + if err != nil { + t.Fatal(err) + } + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + t.Fatal(err) + } + defer w.Stop() + if _, err := kv.Create(ctx, "plan-1.0.ace", []byte("x"), jetstream.KeyTTL(time.Hour)); err != nil { + t.Fatal(err) + } + if err := kv.Delete(ctx, "plan-1.0.ace"); err != nil { + t.Fatal(err) + } + if _, err := kv.Create(ctx, "plan-1.0.g14", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { + t.Fatal(err) + } + for { + select { + case e := <-w.Updates(): + if e == nil || e.Operation() == jetstream.KeyValuePut { + continue + } + if e.Key() == "plan-1.0.ace" && e.Operation() != jetstream.KeyValueDelete { + t.Fatalf("a report's delete was heard as %v", e.Operation()) + } + if e.Key() == "plan-1.0.g14" { + if e.Operation() != jetstream.KeyValuePurge { + t.Fatalf("an expiry was heard as %v, not a purge", e.Operation()) + } + return + } + case <-ctx.Done(): + t.Fatal("no watcher heard the key expire") + } + } +} + // **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/internal/broker/nats.go b/internal/broker/nats.go index e12966bc..396a353c 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -378,6 +378,11 @@ var DegradableGrants = []DegradableGrant{ Reason: "refused: the controller closes a batch's window by a timer of its own for that batch, as the " + "window setting says (ADR 0407 decision 1, issue 973)", }, + { + Subject: "$KV." + RolloutSilentBucket + ".>", + Reason: "refused: the controller advances a rollout on the machines' reports alone, and a machine never heard " + + "from is found when the plan is next looked at (ADR 0406 §5, issue 974)", + }, } // perMachineEvents are a node-scoped seat's events about the holder itself, whose last token is the diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index eb556826..6178d798 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -24,7 +24,7 @@ accounts { jetstream: enabled users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 7e50b9ce..1b03593f 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -117,11 +117,13 @@ var WritersTable = []WriterRow{ Subjects: []string{"mesh.control.*.report", "mesh.control.*.health", "mesh.control.*.cli"}, Writes: ownMachine}, {State: "the controller lease", Writer: "the controller instance holding it", KeptIn: "key-value " + LeaseBucket, Others: "a candidate waits", Subjects: kvOf(LeaseBucket), Writes: isController}, - // And the clock of an open batch's merge window, a plan before it is one (novox/hq ADR 0407 decision 1): keys that - // expire in a bucket of their own, written by the lease holder. + // And the clock of an open batch's merge window, a plan before it is one (novox/hq ADR 0407 decision 1), and the + // wait of a rollout for each machine it sent (ADR 0406 §5, issue 974): keys that expire in buckets of their own, + // written by the lease holder. {State: "plans and their tiers", Writer: "controller (lease holder), compare-and-set on the plan's revision", - KeptIn: "the controller's store; a batch's merge window in key-value " + MergeWindowBucket, Others: "read through plans", - Subjects: kvOf(MergeWindowBucket), Writes: isController}, + KeptIn: "the controller's store; a batch's merge window in key-value " + MergeWindowBucket + + ", a rollout's wait for a machine's report in key-value " + RolloutSilentBucket, Others: "read through plans", + Subjects: append(kvOf(MergeWindowBucket), kvOf(RolloutSilentBucket)...), Writes: isController}, {State: "conditions", Writer: "controller", KeptIn: "key-value " + ConditionsBucket + " (and its history, " + ConditionHistoryBucket + "); how often each was observed, when last and its newest evidence in the " + "controller's store, so a look that changes nothing writes no bucket", Others: "raise or clear only through observations the controller reads",