diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index ddab63c..aeb2101 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -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) } diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index a268cca..4bd3389 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -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()