The node composes again only when what it composes from changed (issue 774) #187

Merged
mesh-tools merged 1 commits from fix/compose-only-on-change into main 2026-10-31 01:07:23 +00:00
3 changed files with 97 additions and 52 deletions
+61 -2
View File
@@ -19,6 +19,7 @@ import (
"encoding/json"
"errors"
"fmt"
"reflect"
"slices"
"sort"
"strings"
@@ -399,8 +400,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
case ins[t.in].changed != nil:
dirty = ins[t.in].changed(c, t.u) || dirty
default:
ins[t.in].take(c, t.u)
dirty = true
dirty = c.takeChanged(ins[t.in], t.u) || dirty
}
if len(merged) > 0 {
continue
@@ -468,6 +468,65 @@ func (c *Composer) caughtSent() bool {
return sent
}
// takeChanged takes u through in.take and says whether the entry it names changed: a want, a placement, a module or a
// config value written again with the value it held changes nothing composed, so it makes no pass (issue 774,
// 2026-10-31: every put under want.<node>.> made a pass, and each pass asked the vault for the node's own secrets).
func (c *Composer) takeChanged(in input, u agent.Update) bool {
before, had, known := c.entry(in.bucket, u.Key)
in.take(c, u)
if !known {
return true
}
after, has, _ := c.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 (c *Composer) entry(bucket, key string) (v any, has, known bool) {
c.mu.Lock()
defer c.mu.Unlock()
switch bucket {
case delivery.AssignerWantsBucket:
// A want is composed by its fingerprint alone (fingerprints); its words, rule and newest change nothing composed.
module := agent.NodeModule
if _, err := delivery.ParseWantNodeKey(key); err != nil {
node, m, err := delivery.ParseWantKey(key)
if err != nil || node != c.env.Node {
return nil, false, true
}
module = m
}
w, ok := c.wants[module]
return w.FP, ok, true
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
}
func (c *Composer) takeWant(u agent.Update) {
c.mu.Lock()
defer c.mu.Unlock()
+35 -1
View File
@@ -546,6 +546,40 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass)
}
}
// Issue 774, 2026-10-31: every put under want.n1.> made a pass, and each pass asked the vault for the node's own
// secrets. A key written again with what it held, or a want whose words moved and not its build, makes none.
worded, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true, Words: "lemurs is held at aaaaaaaaaaaa"})
elsewhere, _ := delivery.EncodeWant(delivery.Want{FP: delivery.SourcePrefix2 + strings.Repeat("e", 64), Self: true})
for _, row := range []struct {
name, bucket, key string
value []byte
pass bool
}{
{"lemurs's want again", delivery.AssignerWantsBucket, "want.n1.lemurs", want, false},
{"the node's own want again", delivery.AssignerWantsBucket, "want.n1.node", nodeWant, false},
{"lemurs's want with other words", delivery.AssignerWantsBucket, "want.n1.lemurs", worded, false},
{"the placement again", delivery.PlacementBucket, "placement.n1", place, false},
{"lemurs's module again", delivery.ModulesBucket, "lemurs", mod, false},
{"lemurs's want at another build", delivery.AssignerWantsBucket, "want.n1.lemurs", elsewhere, true},
{"lemurs's want back at its build", delivery.AssignerWantsBucket, "want.n1.lemurs", want, true},
} {
before := passes()
b.put(t, row.bucket, row.key, row.value)
if !row.pass {
quiet(t, out)
} else {
for done := false; !done; {
select {
case <-out:
case <-time.After(200 * time.Millisecond):
done = true
}
}
}
if got := passes() > before; got != row.pass {
t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass)
}
}
// And today's sequence whole: tool runs at its old build; its want and the trunk move to a build not yet done, and
// the node holds tool, keeping what runs, naming why; the build lands, and with no other change tool is handed at it.
{
@@ -629,7 +663,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
b.mu.Lock()
n := 0
for _, s := range b.said {
if strings.Contains(s, "lemurs is wanted at") {
if strings.Contains(s, "lemurs is wanted at "+gone) {
n++
}
}
+1 -49
View File
@@ -4,7 +4,6 @@ import (
"bytes"
"context"
"fmt"
"reflect"
"sort"
"strings"
"sync"
@@ -231,7 +230,7 @@ func (s *Shadow) Run(ctx context.Context) error {
case t.in == 6:
changed = s.takeWant(t.u)
default:
changed = s.takeChanged(ins[t.in], t.u)
changed = c.takeChanged(ins[t.in], t.u)
}
if changed {
if t.in != 0 {
@@ -264,53 +263,6 @@ func (s *Shadow) Run(ctx context.Context) error {
}
}
// 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 proposal's keys' revisions, by node, as one word.
func revsOf(keys map[string]candidateKey) string {
nodes := make([]string, 0, len(keys))