From e40b1e70db7a1d0dd15a3b49a832ae3c8187986f Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:27:43 +0100 Subject: [PATCH] Follow config from a durable consumer, compose again only what changed since, and hold the node only for a write (issue 1945, ADR 0545, review) The start composed every kept composition of every module on every node, each a whole node composed in memory under the node's lock: 10 to 50 minutes at a start, holding take-ins back. Now the writes of mesh-config_resolved are read from the durable consumer mesh-controller-config-follow, acknowledged once the change is composed, so a restart composes again only what was written since the last change handled, and a first start nothing it did not hear. The node's lock is taken for each write alone. A consumer that cannot be made or read raises config-not-followed and is tried again a minute later, cleared once it reads. The controller's account is granted that consumer's acknowledgements where mesh-config serves, as its verb is. Research 072: milestone 2, the composer reacting to the change of an input. --- cmd/mesh-controller/composed_start.go | 8 - cmd/mesh-controller/kept_compositions.go | 198 +++++++++++++++-------- cmd/mesh-controller/plain_words.go | 7 + internal/broker/composed.go | 4 + internal/broker/nats.go | 5 + 5 files changed, 151 insertions(+), 71 deletions(-) diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go index 7f3fb165..698a0ff5 100644 --- a/cmd/mesh-controller/composed_start.go +++ b/cmd/mesh-controller/composed_start.go @@ -112,14 +112,6 @@ func composeAtStart(ctx context.Context, open *stores) { if err := composeReplacedController(ctx, open); err != nil { fmt.Printf("start: the controller this one replaced is not composed: %v\n", err) } - // And every kept composition with today's inputs, once: a change of config while no controller heard it (ADR 0545). - // One reading of the inputs for every node. - ctx = withInputCache(ctx) - for _, node := range names { - if _, err := followKept(ctx, open, node, nil); err != nil { - fmt.Printf("start: %s's kept compositions are not composed again: %v\n", node, err) - } - } if _, err := retireDeclarations(ctx, open); err != nil { fmt.Printf("start: whether the declarations' stream may go cannot be read: %v\n", err) } diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go index f7f188c1..8258ff45 100644 --- a/cmd/mesh-controller/kept_compositions.go +++ b/cmd/mesh-controller/kept_compositions.go @@ -14,6 +14,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) @@ -23,8 +24,9 @@ import ( // 2026-10-25 a want back to an earlier mesh-build-go on g14 applied its capacity of 0, the value from before it was // set. Now a write of mesh-config's resolved values for a module on a node, heard from that bucket, composes the node // and then every kept fingerprint of that module on that node again, each from its own build's manifest with today's -// inputs, writing only those whose composition differs (writeComposed). A starting controller does it once for every -// node, for a change made while no controller heard it. +// inputs, writing only those whose composition differs (writeComposed). The writes are read from a durable consumer, +// acknowledged once handled, so a change made while no controller ran is delivered to the next one, and a start +// composes again only what was written since the last write handled, never every kept composition. // configTouched is the node and module a key of mesh-config_resolved (`..`) names. func configTouched(key string) (node, module string, ok bool) { @@ -92,8 +94,6 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri if err != nil { return nil, err } - unlock := lockComposing(node) - defer unlock() at, err := openComposedPlace(ctx, js, bus.Conn(), node) if err != nil { return nil, err @@ -160,7 +160,11 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri continue } key, _ := sdkdelivery.ComposedKey(node, k.module, k.fp) + // One write on the node's keys at a time in this process, held only for the write: composing in memory holds + // nothing, so a sweep never keeps a take-in from the node (the review of ADR 0545). + unlock := lockComposing(node) w, err := writeComposed(ctx, at, key, c) + unlock() if err != nil { errs = append(errs, err) continue @@ -175,84 +179,152 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri return wrote, errors.Join(errs...) } -// configFollowGather is how long the config watch waits after a write for more before it composes: mesh-config writes +// configFollowGather is how long the config consumer waits after a write for more before it composes: mesh-config writes // a module's keys one by one. var configFollowGather = time.Second -// watchConfigFollow hears every write of mesh-config's resolved values and, for each node and module it touches, -// composes the node and the module's kept compositions there again, until ctx ends (ADR 0545). A bucket not there is -// said: a change is then composed by the next act that touches the node and the next start. +// configFollowConsumer is the controller's durable consumer of mesh-config's resolved values (ADR 0545 §1, §3). +const configFollowConsumer = broker.ConfigFollowConsumer + +// configFollowConfig is that consumer: each write from when it is made (a first start composes again nothing it did not +// hear), acknowledged after the write is handled, so what a stop cut is delivered to the next controller. +func configFollowConfig() jetstream.ConsumerConfig { + return jetstream.ConsumerConfig{Durable: configFollowConsumer, FilterSubject: "$KV." + broker.ConfigResolvedBucket + ".>", + DeliverPolicy: jetstream.DeliverNewPolicy, AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, + MaxAckPending: takeInPending, Description: "the controller's following of config into kept compositions (novox/hq ADR 0545)"} +} + +// kindConfigNotFollowed is the condition of a config consumer that cannot be read: a change of config composes nothing +// until it is. +const kindConfigNotFollowed = "config-not-followed" + +func configNotFollowedObservation(err error) conditions.Observation { + return conditions.Observation{Scope: conditions.ScopeCore, ID: "config-follow", Token: kindConfigNotFollowed, + Kind: kindConfigNotFollowed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "config", + Summary: fmt.Sprintf("mesh-config's resolved values are not heard, so a change of config composes nothing: %v", err)} +} + +// watchConfigFollow reads the writes of mesh-config's resolved values from the controller's durable consumer and, for +// each node and module they touch, composes the node and the module's kept compositions there again, until ctx ends +// (ADR 0545). A consumer that cannot be made or read raises config-not-followed and is tried again a minute later; it is +// cleared once it is read. func watchConfigFollow(ctx context.Context, open *stores) { - bus, err := aBus() - if err != nil { - fmt.Printf("config: the resolved values are not heard: %v\n", err) - return - } - js, err := jetstream.New(bus.Conn()) - if err != nil { - bus.Close() - fmt.Printf("config: the resolved values are not heard: %v\n", err) - return - } - kv, err := js.KeyValue(ctx, broker.ConfigResolvedBucket) - if err != nil { - bus.Close() - fmt.Printf("config: %s is not heard, so a change of config composes nothing until the next act: %v\n", - broker.ConfigResolvedBucket, err) - return - } - w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) - if err != nil { - bus.Close() - fmt.Printf("config: %s is not heard: %v\n", broker.ConfigResolvedBucket, err) - return - } goServing(ctx, func() { - defer bus.Close() - defer func() { _ = w.Stop() }() - touched := map[string]map[string]bool{} - var gather <-chan time.Time for { + err := followConfigOnce(ctx, open) + if ctx.Err() != nil { + return + } + fmt.Printf("config: %v; tried again in a minute\n", err) + if keeper := conditionsFrom; keeper != nil { + _, _ = keeper.Observe(ctx, configNotFollowedObservation(err)) + } select { case <-ctx.Done(): return - case e, ok := <-w.Updates(): - if !ok { - if ctx.Err() == nil { - fmt.Printf("config: the watch of %s ended; a change is composed by the next start\n", broker.ConfigResolvedBucket) - } - return + case <-time.After(time.Minute): + } + } + }) +} + +// followConfigOnce reads the consumer until it fails or ctx ends. +func followConfigOnce(ctx context.Context, open *stores) error { + bus, err := aBus() + if err != nil { + return err + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return err + } + c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+broker.ConfigResolvedBucket, configFollowConfig()) + if err != nil { + return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err) + } + it, err := c.Messages(jetstream.PullMaxMessages(startTakeIns)) + if err != nil { + return fmt.Errorf("the consumer of %s: %w", broker.ConfigResolvedBucket, err) + } + defer it.Stop() + if keeper := conditionsFrom; keeper != nil { + _, _ = keeper.Clear(ctx, configNotFollowedObservation(nil).Key(), "mesh-config's resolved values are heard") + } + type next struct { + m jetstream.Msg + err error + } + got := make(chan next) + go func() { + for { + m, err := it.Next() + select { + case got <- next{m, err}: + case <-ctx.Done(): + if m != nil { + _ = m.Nak() } - if e == nil { - continue - } - node, module, ok := configTouched(e.Key()) - if !ok { - continue + return + } + if err != nil { + return + } + } + }() + var held []jetstream.Msg + touched := map[string]map[string]bool{} + var gather <-chan time.Time + for { + select { + case <-ctx.Done(): + for _, m := range held { + _ = m.Nak() // delivered at once to the next controller + } + return nil + case n := <-got: + if n.err != nil { + for _, m := range held { + _ = m.Nak() } + return fmt.Errorf("the consumer of %s stopped: %w", broker.ConfigResolvedBucket, n.err) + } + held = append(held, n.m) + key := strings.TrimPrefix(n.m.Subject(), "$KV."+broker.ConfigResolvedBucket+".") + if node, module, ok := configTouched(key); ok { if touched[node] == nil { touched[node] = map[string]bool{} } touched[node][module] = true - if gather == nil { - gather = time.After(configFollowGather) + } + if gather == nil { + gather = time.After(configFollowGather) + } + case <-gather: + gather = nil + now, msgs := touched, held + touched, held = map[string]map[string]bool{}, nil + forgetResolved() // the next composition reads the values just written, not a read of the last 5 s + failed := false + for node, modules := range now { + if err := composerFor(open).compose(ctx, []string{node}, false); err != nil { + fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err) + failed = true } - case <-gather: - gather = nil - now := touched - touched = map[string]map[string]bool{} - forgetResolved() // the next composition reads the values just written, not a read of the last 5 s - for node, modules := range now { - if err := composerFor(open).compose(ctx, []string{node}, false); err != nil { - fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err) - } - if _, err := followKept(ctx, open, node, modules); err != nil { - fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err) - } + if _, err := followKept(ctx, open, node, modules); err != nil { + fmt.Printf("config: %s's kept compositions are not composed again: %v\n", node, err) + failed = true + } + } + for _, m := range msgs { + if failed { + _ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again + } else { + _ = m.Ack() } } } - }) + } } // forgetResolved drops the last read of mesh-config's resolved values. diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index c1ebdf54..e8519748 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -84,6 +84,13 @@ var plainWordings = map[string]func(conditions.Observation) words{ "builds announced there wait until the controller starts again. The details say which and why.", Resolved: "the mesh reads where new builds are announced again"} }), + // Config not followed into the compositions (kept_compositions.go, novox/hq ADR 0545). + kindConfigNotFollowed: worded(func(o conditions.Observation) words { + return words{Headline: "Changes of settings are not reaching the machines", + Explanation: "The mesh could not read the settings as they change, so a changed setting is not prepared for " + + "the machines until it can. It tries again every minute; the details say why.", + Resolved: "changes of settings reach the machines again"} + }), // Machines and their node-engine. "silent": worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") diff --git a/internal/broker/composed.go b/internal/broker/composed.go index b950e058..b7ff58a1 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -190,3 +190,7 @@ const TakeInConsumer = "mesh-controller-take-in" // TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's // builds (each build's key). var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket} + +// ConfigFollowConsumer is the durable consumer the controller follows mesh-config's resolved values with, on that +// bucket's stream (novox/hq ADR 0545). +const ConfigFollowConsumer = "mesh-controller-config-follow" diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 734bffc1..66f66698 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -1273,6 +1273,11 @@ func PermissionsFor(p Principal) (Permissions, error) { pub = append(pub, "mesh.seat."+v.Seat+"."+CallKind+"."+v.Verb+"."+CallerToken(p.Username())) } } + // And its consumer of mesh-config's resolved values, acknowledged once a change is composed (novox/hq ADR 0545): + // where mesh-config serves, as its verb, so genesis grants nothing of mesh-config. + if len(p.AsksTheConfig) > 0 { + pub = append(pub, "$JS.ACK.KV_"+ConfigResolvedBucket+"."+ConfigFollowConsumer+".>") + } } sort.Strings(pub)