From ff3cc47027e04e7d558259447fdf6f9620f9a552 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:43:38 +0100 Subject: [PATCH] Let the kept compositions' pass end at the stop (issue 1945, ADR 0545, review 3) followKept went on composing every kept key, and the config consumer every touched node, after the controller was told to stop. Both now check the stop before each key and each node; the writes of an unfinished batch are refused at once, so the next controller takes them. Research 072: milestone 2, the composer reacting to the change of an input. --- cmd/mesh-controller/kept_compositions.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/cmd/mesh-controller/kept_compositions.go b/cmd/mesh-controller/kept_compositions.go index 87aabfed..7dda518b 100644 --- a/cmd/mesh-controller/kept_compositions.go +++ b/cmd/mesh-controller/kept_compositions.go @@ -137,6 +137,9 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri var errs []error builds := map[string][]inventory.Build{} for _, k := range kept { + if ctx.Err() != nil { + return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's + } if _, read := builds[k.module]; !read { if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil { errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err)) @@ -311,6 +314,10 @@ func followConfigOnce(ctx context.Context, open *stores) error { 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 ctx.Err() != nil { + failed = true // stopped: the writes are refused below and delivered to the next controller + break + } 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 @@ -321,7 +328,9 @@ func followConfigOnce(ctx context.Context, open *stores) error { } } for _, m := range msgs { - if failed { + if failed && ctx.Err() != nil { + _ = m.Nak() // stopped: delivered at once to the next controller + } else if failed { _ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again } else { acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)