From 4182f8fbfb5e2204da5e5e1311d37d4777420c0c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 12:02:57 +0100 Subject: [PATCH] The node answers a placement candidate revision once: a module's have whose health words alone moved, or a key written again with its value, no longer composes every group's answer again (issue 1126) On shanks probe/1126-seed-7 was answered at revision 84 every 13-17 s: each look rewrote the haves of its unhealthy modules with new words, and the shadow took any write of its own module haves, placement, modules or config values as a change. Each answer loaded the bus, the health tools timed out, the words moved again. Now only what an answer is composed from wakes it (what runs, folded modules, a changed value), and each group is answered again only when its keys' revisions or the node's inputs moved. --- cmd/mesh-host/nodeagent_integration_test.go | 58 +++++++++ internal/selfcompose/selfcompose_test.go | 59 +++++++++ internal/selfcompose/shadow.go | 127 +++++++++++++++++--- 3 files changed, 229 insertions(+), 15 deletions(-) diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 089ae2b..4ac6307 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -403,9 +403,67 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { hkey, place, fp := st.hkey, st.place, st.fp within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" }) h8rest(t, b, resolved, candidates, hkey, place, fp) + t.Run("one answer per candidate revision", func(t *testing.T) { h8once(t, b, resolved, candidates, place) }) h8back(t, b, placement, resolved, candidates, hkey) } +// h8once is issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, its bus flooded): +// a candidate revision is answered once. The node's own answer written, a module's have whose health words alone move +// (each look rewrites them) and a config value written again with the value it held answer nothing again; the +// candidate's next revision is answered once. +func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []byte) { + answering := func() int { + b.mu.Lock() + defer b.mu.Unlock() + n := 0 + for _, l := range b.said { + if strings.HasPrefix(l, "answering placement candidate feat/x ") { + n++ + } + } + return n + } + time.Sleep(time.Second) + settled := answering() + if settled == 0 { + t.Fatal("feat/x was never answered") + } + hkey, _ := delivery.HaveKey(itNode, "web") + for _, detail := range []string{"its tool web_health: no link to the bus is open", "its tool web_health: no answer within 10s"} { + h := b.have("web") + h.Health, h.Detail = delivery.HealthFailed, detail + raw, err := delivery.EncodeHave(h) + if err != nil { + t.Fatal(err) + } + if _, err := b.haves.Put(hkey, raw); err != nil { + t.Fatal(err) + } + if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil { + t.Fatal(err) + } + } + time.Sleep(1500 * time.Millisecond) + if got := answering(); got != settled { + t.Errorf("feat/x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled) + } + same, _ := delivery.DecodePlacement(place) + craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x", + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1b"}, Placement: &same}) + if err != nil { + t.Fatal(err) + } + ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) + if _, err := candidates.Put(ckey, craw); err != nil { + t.Fatal(err) + } + within(t, 10*time.Second, "the candidate's next revision was not answered", func() bool { return answering() > settled }) + time.Sleep(1500 * time.Millisecond) + if got := answering(); got != settled+1 { + t.Errorf("the candidate's next revision was answered %d time(s), want 1", got-settled) + } +} + // h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with // the controller's compositions at its wants (here none: refused, naming web), and the placement key deleted returns the agent to them without a // restart, said in the journal and the node's have, keeping what runs. diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index b106ca4..84c818b 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -721,6 +721,65 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Errorf("answers handed again for nothing changed: %+v", a) case <-time.After(100 * time.Millisecond): } + // Issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s): a candidate revision is + // answered once. A module's have whose health or words alone move (each look rewrites them), and a placement, a + // module or a config value written again with the value it held, answer nothing again; a new revision answers its + // group alone, and what runs moving answers every group once. + answering := func() int { + readOnly.mu.Lock() + defer readOnly.mu.Unlock() + n := 0 + for _, l := range readOnly.said { + if strings.HasPrefix(l, "answering placement candidate ") { + n++ + } + } + return n + } + drain := func() { + for { + select { + case answers = <-answered: + case <-time.After(150 * time.Millisecond): + return + } + } + } + lemursHave := func(detail string, rev uint64) { + raw, err := delivery.EncodeHave(delivery.Have{FP: fp, Composed: lemursDigest, Phase: delivery.PhaseRunning, + Health: delivery.HealthFailed, Detail: detail}) + if err != nil { + t.Fatal(err) + } + hk, _ := delivery.HaveKey("n1", "lemurs") + sb.put(delivery.HaveBucket, hk, raw, rev) + } + lemursHave("its tool lemurs_health: no link to the bus is open", 10) + drain() + settled := answering() + lemursHave("its tool lemurs_health: no answer within 10s", 11) + lemursHave("its tool lemurs_health: no answer within 20s", 12) + sb.put(delivery.PlacementBucket, "placement.n1", tp, 2) + lraw, _ := delivery.EncodeModule(mods["lemurs"]) + sb.put(delivery.ModulesBucket, "lemurs", lraw, 2) + drain() + if got := answering(); got != settled { + t.Errorf("answered %d more time(s) with no candidate revision and no input of the node changed", got-settled) + } + candidate("same", 8, "lemurs") + waitFor("the new revision answered", func(a map[string]delivery.CandidateAnswer) bool { return a["same"].Revision == 8 }) + drain() + if got := answering(); got != settled+1 { + t.Errorf("a new revision of one group answered %d group(s), want 1", got-settled) + } + settled = answering() + hk, _ := delivery.HaveKey("n1", "lemurs") + moved, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: delivery.PhaseRunning}) + sb.put(delivery.HaveBucket, hk, moved, 13) + drain() + if got := answering(); got != settled+len(answers) { + t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers)) + } k, _ := delivery.PlacementCandidateKey("same", "n1") sb.del(delivery.PlacementCandidatesBucket, k) waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 }) diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index bcdc263..cec080e 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "reflect" "sort" "strings" "sync" @@ -39,6 +40,18 @@ type Shadow struct { // ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy // from the declaration kept before the switch (issue 1126, probe #532 on shanks). ran map[string]string + // inputs counts the changes of the node's own inputs (everything but the candidates), and answered is, by group, the + // candidate revisions and the count each answer was composed at: a group is answered again only when either moved + // (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, woken by its own modules' + // haves whose health words changed at every look). + inputs uint64 + answered map[string]answeredAt +} + +// answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the count of input changes. +type answeredAt struct { + revs string + inputs uint64 } // candidateKey is one candidate key as last read: its value and revision. @@ -94,6 +107,7 @@ func (s *Shadow) Run(ctx context.Context) error { s.candidates = map[string]map[string]candidateKey{} s.wants = map[string]delivery.Want{} s.ran = map[string]string{} + s.answered = map[string]answeredAt{} s.c = New(s.c.env) s.mu.Unlock() c := s.c @@ -153,21 +167,29 @@ func (s *Shadow) Run(ctx context.Context) error { if !t.ok { return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket) } + // Only a change of what an answer is composed from answers again: a key written again with the value it held, + // this node's own key (its answers), or a module's have whose health or words moved but not what it runs, is + // none (issue 1126, 2026-10-28: each look rewrote shanks' unhealthy modules' haves, and each woke a full answer). + changed := false switch { case t.u.Caught: caught[t.in] = true case t.in == 0: - dirty = s.takeCandidate(t.u) || dirty + changed = s.takeCandidate(t.u) case t.in == 2: - dirty = s.takeRecord(t.u) || dirty + changed = s.takeRecord(t.u) case t.in == 5: - c.takeModuleHave(t.u) - s.takeRan(t.u) - dirty = !isNodeKey(t.u.Key) || dirty + folded := c.takeModuleHave(t.u) + changed = s.takeRan(t.u) || folded case t.in == 6: - dirty = s.takeWant(t.u) || dirty + changed = s.takeWant(t.u) default: - ins[t.in].take(c, t.u) + changed = s.takeChanged(ins[t.in], t.u) + } + if changed { + if t.in != 0 { + s.inputs++ // the candidates' own changes are their revisions + } dirty = true } if len(merged) > 0 { @@ -187,9 +209,65 @@ func (s *Shadow) Run(ctx context.Context) error { } } -func isNodeKey(key string) bool { - _, _, isNode, err := delivery.ParseHaveKey(key) - return err == nil && isNode +// takeChanged takes u through in.take and says whether the entry it names changed: a placement, a module or a config +// value written again with the value it held changes nothing composed. +func (s *Shadow) takeChanged(in input, u agent.Update) bool { + before, had, known := s.entry(in.bucket, u.Key) + in.take(s.c, u) + if !known { + return true + } + after, has, _ := s.entry(in.bucket, u.Key) + return had != has || !reflect.DeepEqual(before, after) +} + +// entry is what the composer holds of one key of bucket, whether it holds it, and whether the bucket is one it knows. +func (s *Shadow) entry(bucket, key string) (v any, has, known bool) { + c := s.c + c.mu.Lock() + defer c.mu.Unlock() + switch bucket { + case delivery.PlacementBucket: + node, err := delivery.ParsePlacementKey(key) + if err != nil { + return nil, false, true + } + p, ok := c.placements[node] + return p, ok, true + case delivery.ModulesBucket: + name, err := delivery.ParseModulesKey(key) + if err != nil { + return nil, false, true + } + m, ok := c.modules[name] + return m, ok, true + case ResolvedBucket: + rest, ok := strings.CutPrefix(key, c.env.Node+".") + if !ok { + return nil, false, true + } + module, k, ok := strings.Cut(rest, ".") + if !ok { + return nil, false, true + } + val, ok := c.values[module][k] + return val, ok, true + } + return nil, false, false +} + +// revsOf is a group's keys' revisions, by node, as one word. +func revsOf(keys map[string]candidateKey) string { + nodes := make([]string, 0, len(keys)) + for n := range keys { + nodes = append(nodes, n) + } + sort.Strings(nodes) + var b strings.Builder + for _, n := range nodes { + fmt.Fprintf(&b, "%s=%d;", n, keys[n].rev) + } + return b.String() } // takeCandidate takes one candidate key, and says whether it changed anything. @@ -289,15 +367,22 @@ func (s *Shadow) answerAll(ctx context.Context) { type listed struct { group string rev uint64 + at answeredAt } var groups []listed for g, keys := range s.candidates { - l := listed{group: g} + l := listed{group: g, at: answeredAt{revs: revsOf(keys), inputs: s.inputs}} for _, k := range keys { l.rev = max(l.rev, k.rev) } groups = append(groups, l) } + // A group whose keys' revisions and the node's inputs are those its answer stands at is not answered again. + for g := range s.answered { + if _, ok := s.candidates[g]; !ok { + delete(s.answered, g) + } + } // Each group's last answer stands until it is answered again; a group whose keys went is dropped. next := map[string]delivery.CandidateAnswer{} for _, l := range groups { @@ -342,6 +427,12 @@ func (s *Shadow) answerAll(ctx context.Context) { return } group := l.group + s.mu.Lock() + at, done := s.answered[group] + s.mu.Unlock() + if done && at == l.at { + continue + } began := time.Now() c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev)) stop := make(chan struct{}) @@ -376,6 +467,9 @@ func (s *Shadow) answerAll(ctx context.Context) { delete(next, group) c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group)) } + s.mu.Lock() + s.answered[group] = l.at + s.mu.Unlock() hand() } hand() // a group whose keys went, when no group is left to answer @@ -470,20 +564,23 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid return asHavesSay(a, ran), true } -// takeRan keeps the digest one of this node's module haves says runs. -func (s *Shadow) takeRan(u agent.Update) { +// takeRan keeps the digest one of this node's module haves says runs, and says whether it changed: a have whose health +// or words alone moved runs the same. +func (s *Shadow) takeRan(u agent.Update) bool { node, module, isNode, err := delivery.ParseHaveKey(u.Key) if err != nil || isNode || node != s.c.env.Node { - return + return false } s.mu.Lock() defer s.mu.Unlock() + before, had := s.ran[module] h, err := delivery.DecodeHave(u.Value) if u.Deleted || err != nil || h.Composed == "" || h.Phase == delivery.PhaseRemoved { delete(s.ran, module) - return + return had } s.ran[module] = h.Composed + return !had || before != h.Composed } // asHavesSay is a with each module that differs only from the copy the agent keeps of the declaration before the switch