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

Merge branch 'fix/1126-answer-once' into fix/1126-seed-same-3

Browse Source
This commit is contained in:
jschoubben
2026-10-28 12:11:50 +01:00
parent faf6a23d91 cd1d2ff447
commit 16df7d0b7a
9 changed files with 534 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/selfcompose/selfcompose.go
+104 -9
View File
@@ -20,6 +20,7 @@ import (
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -66,7 +67,8 @@ type Env struct {
// before the agent applied anything, and after the switch once the kept declaration is old. By slot
// (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused
// only within its own module.
Held func(module string) (slots delivery.Seals, legacy []string)
// An error is a have the bus did not answer for, never one absent: the pass is then tried again (BusFailed).
Held func(module string) (slots delivery.Seals, legacy []string, err error)
// Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built
// manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose.
Fetch func(ctx context.Context, url string) ([]byte, error)
@@ -102,6 +104,25 @@ type Composer struct {
doing string
caught bool
rev uint64
// missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one:
// a read that failed on the bus is tried again after a bound, never kept as an answer (issue 1126, 2026-10-28).
missed atomic.Uint64
// owed is set when a pass refused for a bus that did not answer: the composer composes again after a bound.
owed atomic.Bool
}
// note counts err when it is the bus not answering, and returns it.
func (c *Composer) note(err error) error {
if BusFailed(err) {
c.missed.Add(1)
}
return err
}
// get is env.Get, its failures noted.
func (c *Composer) get(ctx context.Context, bucket, key string) ([]byte, error) {
raw, err := c.env.Get(ctx, bucket, key)
return raw, c.note(err)
}
// New is a composer for one node.
@@ -204,11 +225,18 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
caught := make([]bool, len(ins))
ready := false
dirty := false
// A pass refused for a bus that did not answer is made again after a bound (laterAgain), not kept until another
// input moves (issue 1126, 2026-10-28).
var again <-chan time.Time
tries := 0
for {
// Drain whatever came, then compose once for all of it.
select {
case <-ctx.Done():
return
case <-again:
again = nil
dirty = true
case t := <-merged:
if !t.ok {
c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket))
@@ -247,6 +275,14 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
return
}
}
if c.owed.Load() {
wait := LaterWait(tries)
tries++
c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
again = time.After(wait)
} else {
tries, again = 0, nil
}
if !c.caughtSent() {
select {
case out <- agent.Update{Caught: true}:
@@ -459,7 +495,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
running = c.env.Running()
}
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}
missed := c.missed.Load()
compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
err = c.missedIn(missed, err)
// A bus that does not answer for a moment (a reload of its accounts pauses it a few seconds) is no refusal: the pass is
// tried again after a growing wait before anything is marked (issue 1126).
for try := 0; err != nil && Transient(err) && ctx.Err() == nil && try < len(composeAgain); try++ {
@@ -469,7 +507,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
case <-time.After(composeAgain[try]):
}
if ctx.Err() == nil {
missed = c.missed.Load()
compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
err = c.missedIn(missed, err)
}
}
if ctx.Err() != nil {
@@ -481,14 +521,25 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
// Said before the agent marks the modules: the marks read each have first, which a paused bus delays.
c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
c.refused(fmt.Sprintf("%v", err))
// A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28).
c.owed.Store(BusFailed(err))
return nil
}
c.refused("")
c.sayOnce("")
// The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs.
missed = c.missed.Load()
for key, comp := range compositions {
compositions[key] = c.keepRunning(comp, running)
}
if c.missed.Load() > missed {
// A have of what runs the bus did not answer for: its boxes would be the vault's, sealed again, and the
// composition another; nothing is handed, and the pass is made again after a bound (issue 1126, 2026-10-28).
c.sayOnce("the node composes nothing new for now, and keeps running what it runs: the have of what it runs did not answer on the bus")
c.owed.Store(true)
return nil
}
c.owed.Store(false)
c.refused("")
c.sayOnce("")
c.mu.Lock()
defer c.mu.Unlock()
var out []agent.Update
@@ -557,7 +608,8 @@ func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Compos
kept := delivery.Seals{}
var legacy []string
if c.env.Held != nil {
slots, old := c.env.Held(module)
slots, old, err := c.env.Held(module)
c.note(err)
for slot, box := range slots {
kept[slot] = box
}
@@ -687,7 +739,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
return nil, err
}
defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")()
value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key)
value, err := r.c.get(ctx, delivery.ManifestsBucket, key)
if err != nil {
return nil, err
}
@@ -710,7 +762,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder
return nil, err
}
defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")()
raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
raw, err := r.c.get(ctx, delivery.BuilderBuildsBucket, key)
if err != nil || raw == nil {
return nil, err
}
@@ -740,7 +792,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error)
fetching, cancel := context.WithTimeout(ctx, archiveWithin)
defer cancel()
body, err := r.c.env.Fetch(fetching, url)
if err != nil {
if r.c.note(err) != nil {
return nil, err
}
sum := sha256.Sum256(body)
@@ -759,7 +811,7 @@ func (r reader) Served(ctx context.Context, node, module string, keys []string)
defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")()
out := map[string]any{}
for _, k := range keys {
raw, err := r.c.env.Get(ctx, ResolvedBucket, node+"."+module+"."+k)
raw, err := r.c.get(ctx, ResolvedBucket, node+"."+module+"."+k)
if err != nil {
return nil, err
}
@@ -797,6 +849,7 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre
}
done := r.c.waiting("the vault's own secrets")
raw, err := r.c.env.Own(ctx, map[string]any{})
r.c.note(err)
done()
if err != nil {
return nil, fmt.Errorf("the vault's own: %w", err)
@@ -853,6 +906,7 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
}
done := r.c.waiting("the vault's credentials")
raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
r.c.note(err)
done()
if err != nil {
return nil, nil, fmt.Errorf("the vault's own: %w", err)
@@ -938,6 +992,46 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
// composeAgain are the waits before a pass that met a bus not answering is tried again.
var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}
// laterAgain are the waits, after a pass or an answer refused for a bus that did not answer past composeAgain, before
// it is made again; the last repeats while the bus does not answer (issue 1126, 2026-10-28: a failed read was kept as
// the answer until another input moved).
var laterAgain = []time.Duration{5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute}
// LaterWait is the wait before the try'th making again (from 0), of a pass, an answer or an ask the bus did not answer.
func LaterWait(try int) time.Duration { return laterAgain[min(try, len(laterAgain)-1)] }
// ErrBusMissed is a pass that met a read the bus did not answer, though the composition was made around it (a have of
// sealed values that could not be read): never handed as made.
var ErrBusMissed = errors.New("a read on the bus did not answer during the pass")
// missedIn is err, or ErrBusMissed when no error came but a read since missed failed on the bus.
func (c *Composer) missedIn(missed uint64, err error) error {
if err == nil && c.missed.Load() > missed {
return ErrBusMissed
}
return err
}
// BusFailed says err is a read the bus did not answer, for any reason but the run's own end: Transient, a read that
// timed out, or no link to the bus. Such a read is tried again after a bound, never kept as an answer; a key absent, a
// value refused or a verdict is an answer (issue 1126, 2026-10-28). Matched by value and by words, as Transient.
func BusFailed(err error) bool {
if err == nil || errors.Is(err, context.Canceled) {
return false
}
if Transient(err) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) ||
errors.Is(err, nats.ErrConnectionClosed) || errors.Is(err, nats.ErrDisconnected) || errors.Is(err, nats.ErrNoServers) {
return true
}
words := err.Error()
for _, w := range []string{"timeout", "deadline exceeded", "no link to the bus", "did not answer", "connection closed"} {
if strings.Contains(words, w) {
return true
}
}
return false
}
// Transient says err is the bus not answering for a moment, as a reload of its accounts makes it for a few seconds:
// JetStream not enabled, or no responders; each fails at once, so a pass tried again costs only its wait. A read that
// timed out is not one: it already waited longer than such a pause, and is most often a read the bus refuses. Matched
@@ -946,7 +1040,8 @@ func Transient(err error) bool {
if err == nil {
return false
}
if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) {
if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) ||
errors.Is(err, ErrBusMissed) {
return true
}
words := err.Error()
internal/selfcompose/selfcompose_test.go
+105 -12
View File
@@ -10,6 +10,7 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
@@ -306,6 +307,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
Transient(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) {
t.Error("Transient misjudges a bus that does not answer, or a refusal")
}
// A read the bus did not answer, timed out or not, is tried again later; the run's own end and a refusal are not.
if !BusFailed(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) || !BusFailed(fmt.Errorf("x: %w", nats.ErrNoResponders)) ||
!BusFailed(fmt.Errorf("the have of web: %w", nats.ErrTimeout)) || BusFailed(fmt.Errorf("x: %w", context.Canceled)) ||
BusFailed(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) || BusFailed(nil) {
t.Error("BusFailed misjudges a read the bus did not answer, the run's end, or a refusal")
}
// Issue 1126, C2: a config value is laid as the controller lays it (catalogue.LaidValue), from the mesh's, the node's
// or the module's own and never from a default or an unset one, both of this node's and of what another node serves:
@@ -428,23 +435,28 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
name, in, held string
slots delivery.Seals
kept bool
fails bool // the have's read the bus did not answer: no composition, made again later (2026-10-28)
}{
{"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true},
{"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false},
{"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true},
{"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false},
{"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false},
{"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true, false},
{"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false, false},
{"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true, false},
{"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false, false},
{"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false, false},
{"the bus does not answer for the have", "keeper", rotatedBox, nil, false, true},
} {
restarted := env
restarted.Running = func() []delivery.Composed { return nil }
restarted.Held = func(module string) (delivery.Seals, []string) {
restarted.Held = func(module string) (delivery.Seals, []string, error) {
if module != row.in {
return nil, nil
return nil, nil, nil
}
if row.fails {
return nil, nil, fmt.Errorf("the have of %s: %w", module, nats.ErrTimeout)
}
if row.slots == nil {
return nil, []string{row.held}
return nil, []string{row.held}, nil
}
return row.slots, nil
return row.slots, nil, nil
}
restarted.Unseal = func(box string) ([]byte, error) {
b.mu.Lock()
@@ -475,6 +487,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
composed = string(u.Value)
}
}
if row.fails {
if composed != "" || !again.owed.Load() {
t.Errorf("%s: composed %v and owes a pass again %v; want nothing handed and a pass owed", row.name, composed != "", again.owed.Load())
}
continue
}
if composed == "" {
t.Fatalf("%s: keeper was not composed", row.name)
}
@@ -497,11 +515,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
}
return nil, fmt.Errorf("not sealed to n1: %s", box)
}
coreEnv.Held = func(module string) (delivery.Seals, []string) {
coreEnv.Held = func(module string) (delivery.Seals, []string, error) {
if module == agent.NodeModule {
return delivery.Seals{"core-box#sealed": rotatedBox}, nil
return delivery.Seals{"core-box#sealed": rotatedBox}, nil, nil
}
return nil, nil
return nil, nil, nil
}
if got := delivery.SealsIn(New(coreEnv).keepRunning(ownComp, nil))["core-box#sealed"]; got != rotatedBox {
t.Errorf("the node's own core box after a restart is %s, want the one its key holds %s", got, rotatedBox)
@@ -606,7 +624,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
senv := readOnly.env()
senv.Watch = sb.watch
var ownFails atomic.Bool // the vault's own on a bus that does not answer (2026-10-28, ace's answer to seed-7)
senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
if ownFails.Load() {
return nil, fmt.Errorf("the vault's own could not be asked: %w", nats.ErrNoResponders)
}
return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
}
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
@@ -755,6 +777,77 @@ 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))
}
// A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved.
laterAgain = []time.Duration{20 * time.Millisecond}
ownFails.Store(true)
candidate("adds", 9, "lemurs", "keeper")
waitFor("the answer met the vault not answering", func(a map[string]delivery.CandidateAnswer) bool {
return a["adds"].Revision == 9 && strings.Contains(a["adds"].Refused, "no responders")
})
ownFails.Store(false)
waitFor("answered again once the bus answers, with nothing else moved", func(a map[string]delivery.CandidateAnswer) bool {
return a["adds"].Revision == 9 && a["adds"].Verdict == delivery.CandidateDiffers
})
drain()
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
+140 -16
View File
@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"fmt"
"reflect"
"sort"
"strings"
"sync"
@@ -39,6 +40,22 @@ 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
// owed says an answer met a read the bus did not answer: it is no answer to keep, and its group is answered again
// after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not
// yet ... no responders" in its answer to seed-7).
owed bool
}
// 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 +111,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
@@ -140,10 +158,15 @@ func (s *Shadow) Run(ctx context.Context) error {
// knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531).
waiting := time.NewTimer(notReadyAfter)
defer waiting.Stop()
var again <-chan time.Time
tries := 0
for {
select {
case <-ctx.Done():
return nil
case <-again:
again = nil
dirty = true
case <-waiting.C:
if !ready {
s.notReady(ctx, ins, caught)
@@ -153,21 +176,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 {
@@ -183,13 +214,77 @@ func (s *Shadow) Run(ctx context.Context) error {
if dirty {
dirty = false
s.answerAll(ctx)
if s.owed {
wait := LaterWait(tries)
tries++
c.env.Say(fmt.Sprintf("this node answers again in %s the placement candidates whose answer met a bus that did not answer", wait))
again = time.After(wait)
} else {
tries, again = 0, nil
}
}
}
}
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 +384,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 {
@@ -305,6 +407,7 @@ func (s *Shadow) answerAll(ctx context.Context) {
next[l.group] = a
}
}
s.owed = false
s.mu.Unlock()
sort.Slice(groups, func(i, j int) bool {
if groups[i].rev != groups[j].rev {
@@ -342,6 +445,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{})
@@ -358,8 +467,10 @@ func (s *Shadow) answerAll(ctx context.Context) {
}
}
}()
missed := c.missed.Load()
a, ok := s.answerGroup(ctx, group)
close(stop)
provisional := c.missed.Load() > missed && ctx.Err() == nil
took := time.Since(began)
if took > slowAfter {
c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second)))
@@ -376,6 +487,16 @@ 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()
if provisional {
// What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
s.owed = true
delete(s.answered, group)
c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", group))
} else {
s.answered[group] = l.at
}
s.mu.Unlock()
hand()
}
hand() // a group whose keys went, when no group is left to answer
@@ -428,7 +549,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
wants[m] = w
}
s.mu.Unlock()
return returning(ctx, c.env.Get, node, wants, running, head, rev), true
return returning(ctx, c.get, node, wants, running, head, rev), true
}
placement := given[node]
@@ -470,20 +591,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