From 11725908ec12213911b007ece197e3c5b3227a0c Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 12:07:57 +0200 Subject: [PATCH] The merge window's bucket: keys that expire, written by the controller (issue 973) --- internal/broker/controller_buckets.go | 29 +++++++++++- internal/broker/controller_buckets_test.go | 52 ++++++++++++++++++++++ internal/broker/nats.go | 5 +++ internal/broker/testdata/composed.conf | 2 +- internal/broker/writers.go | 5 ++- 5 files changed, 89 insertions(+), 4 deletions(-) diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 9c8d841c..1c534d0d 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -38,6 +38,11 @@ var ( // each ask by its id, its options and the actions they stand for, how it ended and whether the // controller acted on its warrant — so a restart neither asks twice nor acts twice. AskedBucket = BucketName(ControllerSeat, "asked") + // MergeWindowBucket holds the clock of every open batch's merge window (novox/hq ADR 0407 decision 1): per + // batch a key `.quiet`, made again by each merge into it and living the window, and a key + // `.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") ) // The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of @@ -79,6 +84,10 @@ const ( MirrorMarkerTTL = 24 * time.Hour ) +// MergeWindowMarkerTTL is how long the marker an expired window key leaves is kept: long enough for a watcher +// that is up to hear it, short enough to leave nothing behind. +const MergeWindowMarkerTTL = 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 @@ -106,13 +115,14 @@ const ( 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 == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || + bucket == MergeWindowBucket } // 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} + AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -209,6 +219,21 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err) } + // **The merge window: keys that expire.** A key's own time to live is the window; the marker an expired key + // leaves is what a watcher hears, kept a minute (ADR 0407 decision 1). + if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: MergeWindowBucket, + Description: "the clock of every open batch's merge window (novox/hq ADR 0407): `.quiet` lives the " + + "window and is made again by each merge, `.most` lives the most; the first to expire closes the " + + "batch; written by the controller alone", + History: 1, + MaxValueSize: 4 << 10, + MaxBytes: 8 << 20, + Storage: jetstream.FileStorage, + LimitMarkerTTL: MergeWindowMarkerTTL, + }); err != nil { + return fmt.Errorf("asserting bucket %s: %w", MergeWindowBucket, 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 12bcc1da..af115301 100644 --- a/internal/broker/controller_buckets_test.go +++ b/internal/broker/controller_buckets_test.go @@ -268,3 +268,55 @@ func TestTheControllerMayAskMeshConfigInItsOwnName(t *testing.T) { } } } + +// **The merge window is a bucket whose keys expire** (novox/hq ADR 0407 decision 1, issue 973): a key made with a +// time to live goes when it is up, and a watcher on the bucket hears it go. The controller may write it, and it is +// no undeclared state. +func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) { + controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(controller.Publish, "$KV."+MergeWindowBucket+".>") { + t.Error("the controller may not write the merge window's bucket") + } + if !IsControllerBucket(MergeWindowBucket) || !slices.Contains(ControllerBuckets(), MergeWindowBucket) { + t.Error("the merge window's 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, MergeWindowBucket) + if err != nil { + t.Fatal(err) + } + w, err := kv.WatchAll(ctx) + if err != nil { + t.Fatal(err) + } + defer w.Stop() + if _, err := kv.Create(ctx, "plan-1.quiet", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil { + t.Fatal(err) + } + for { + select { + case e := <-w.Updates(): + if e != nil && e.Key() == "plan-1.quiet" && e.Operation() != jetstream.KeyValuePut { + return + } + case <-ctx.Done(): + t.Fatal("no watcher heard the key expire") + } + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index ffbca362..f11cdc98 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -368,6 +368,11 @@ var DegradableGrants = []DegradableGrant{ Subject: "mesh.seat." + VerbsTheControllerAsksTheOperatorChannel[0].Seat + ".tool." + VerbsTheControllerAsksTheOperatorChannel[0].Verb, Reason: "refused: the controller raises the secret-given condition as before (ADR 0380 §9)", }, + { + Subject: "$KV." + MergeWindowBucket + ".>", + 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)", + }, } // 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 b7134718..4febcc69 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.>", "$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_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.>", "$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"] } 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 3d760014..7e50b9ce 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -117,8 +117,11 @@ 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. {State: "plans and their tiers", Writer: "controller (lease holder), compare-and-set on the plan's revision", - KeptIn: "the controller's store", Others: "read through plans"}, + KeptIn: "the controller's store; a batch's merge window in key-value " + MergeWindowBucket, Others: "read through plans", + Subjects: kvOf(MergeWindowBucket), 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",