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

Merge pull request 'The node's shadow says which input woke each answer (issue 1126)' (#163) from fix/1126-vault-own into main

Browse Source
This commit was merged in pull request #163.
This commit is contained in:
mesh-tools
2026-10-28 14:20:39 +00:00
parent b4fd2cc210 8bd0b5db15
commit 690aa662a8
2 changed files with 69 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/selfcompose/selfcompose_test.go
+22 -2
View File
@@ -836,6 +836,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
if got := answering(); got != settled+1 {
t.Errorf("a new revision of one group answered %d group(s), want 1", got-settled)
}
// Issue 1126, 2026-10-28 (seed-9's probe): each answer says which input woke it, so a loop is told from inputs moving.
lastSaid := func(prefix string) string {
readOnly.mu.Lock()
defer readOnly.mu.Unlock()
for i := len(readOnly.said) - 1; i >= 0; i-- {
if strings.HasPrefix(readOnly.said[i], prefix) {
return readOnly.said[i]
}
}
return ""
}
if got := lastSaid("answering placement candidate same "); got != "answering placement candidate same (revision 8; woken by the candidate)" {
t.Errorf("a new revision's answer said %q", got)
}
settled = answering()
hk, _ := delivery.HaveKey("n1", "lemurs")
moved, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: delivery.PhaseRunning})
@@ -844,6 +858,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
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))
}
if got := lastSaid("answering placement candidate drops "); got != "answering placement candidate drops (revision 5; woken by haves)" {
t.Errorf("an answer woken by what runs moving said %q", got)
}
// A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved.
setFor(t, &laterAgain, []time.Duration{20 * time.Millisecond})
ownFails.Store(true)
@@ -856,6 +873,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
return a["adds"].Revision == 9 && a["adds"].Verdict == delivery.CandidateDiffers
})
drain()
if got := lastSaid("answering placement candidate adds "); got != "answering placement candidate adds (revision 9; woken by a retry after a read the bus did not answer)" {
t.Errorf("an answer composed again after the bus did not answer said %q", got)
}
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 })
@@ -973,8 +993,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
// Issue 1126, T2: an answer says when it starts and when it ends (and, every minute between, what it waits on).
nmu.Lock()
said := strings.Join(nsaid, "\n")
for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3)",
"answered placement candidate same in", "answering placement candidate older (revision 2)"} {
for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3; woken by its first answer since the shadow began)",
"answered placement candidate same in", "answering placement candidate older (revision 2;"} {
if !strings.Contains(said, line) {
t.Errorf("%q was not said: %q", line, nsaid)
}
internal/selfcompose/shadow.go
+47 -11
View File
@@ -40,22 +40,52 @@ 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 is, by group, the candidate revisions and the input changes (moved) 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).
answered map[string]answeredAt
// moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the groups
// answered again after a read the bus did not answer: what woke each answer is said with it (issue 1126,
// 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving).
moved [len(inputWords)]uint64
retry map[string]bool
// 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.
// answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the changes of each input.
type answeredAt struct {
revs string
inputs uint64
revs string
moved [len(inputWords)]uint64
}
// inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it.
var inputWords = [...]string{"the candidate", "placement", "records", "modules", "values", "haves", "wants"}
// wokenBy says what moved between the answer a group stands at (had, when done) and the one about to be composed (at):
// its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words.
func wokenBy(had answeredAt, done, retry bool, at answeredAt) string {
switch {
case retry:
return "a retry after a read the bus did not answer"
case !done:
return "its first answer since the shadow began"
}
var words []string
if had.revs != at.revs {
words = append(words, inputWords[0])
}
for i := 1; i < len(inputWords); i++ {
if had.moved[i] != at.moved[i] {
words = append(words, inputWords[i])
}
}
if len(words) == 0 {
return "nothing that moved"
}
return strings.Join(words, ", ")
}
// candidateKey is one candidate key as last read: its value and revision.
@@ -112,10 +142,13 @@ func (s *Shadow) Run(ctx context.Context) error {
s.wants = map[string]delivery.Want{}
s.ran = map[string]string{}
s.answered = map[string]answeredAt{}
s.moved = [len(inputWords)]uint64{}
s.retry = map[string]bool{}
s.c = New(s.c.env)
s.mu.Unlock()
c := s.c
node := c.env.Node
// In inputWords' order.
ins := []input{
{delivery.PlacementCandidatesBucket, ">", nil, nil},
{delivery.PlacementBucket, ">", (*Composer).takePlacement, nil},
@@ -197,7 +230,7 @@ func (s *Shadow) Run(ctx context.Context) error {
}
if changed {
if t.in != 0 {
s.inputs++ // the candidates' own changes are their revisions
s.moved[t.in]++ // the candidates' own changes are their revisions
}
dirty = true
}
@@ -388,7 +421,7 @@ func (s *Shadow) answerAll(ctx context.Context) {
}
var groups []listed
for g, keys := range s.candidates {
l := listed{group: g, at: answeredAt{revs: revsOf(keys), inputs: s.inputs}}
l := listed{group: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}}
for _, k := range keys {
l.rev = max(l.rev, k.rev)
}
@@ -447,12 +480,13 @@ func (s *Shadow) answerAll(ctx context.Context) {
group := l.group
s.mu.Lock()
at, done := s.answered[group]
retry := s.retry[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))
c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", group, l.rev, wokenBy(at, done, retry, l.at)))
stop := make(chan struct{})
go func() {
tick := time.NewTicker(progressEvery)
@@ -491,10 +525,12 @@ func (s *Shadow) answerAll(ctx context.Context) {
if provisional {
// What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
s.owed = true
s.retry[group] = 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
delete(s.retry, group)
}
s.mu.Unlock()
hand()