novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

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)

Browse Source 
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.
This commit is contained in:
jschoubben
2026-10-28 12:02:57 +01:00
parent c3a2ffde67
commit 4182f8fbfb
3 changed files with 229 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/selfcompose/selfcompose_test.go
+59
View File
@@ -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 })
internal/selfcompose/shadow.go
+112 -15
View File
@@ -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