The node composes again only when what it composes from changed (issue 774) #187
@@ -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()
|
||||
|
||||
@@ -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++
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user