diff --git a/go.mod b/go.mod index 927562c..afcb4ea 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028000902-4e0c57771e8c github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index 76c4f22..e7cef10 100644 --- a/go.sum +++ b/go.sum @@ -26,6 +26,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024 h1:MnmyspVo git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d h1:DhR7DRKaF0OtA5l/ikIk7gZGeIM1lRz65PUE1DGEkvU= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028000902-4e0c57771e8c h1:4lS82cVTtxls17bRn3551+m2xlZO+0WEtv0a7iME8Ec= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028000902-4e0c57771e8c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 6a3eaa9..4df0ee4 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -692,6 +692,65 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } + // Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in + // a refused answer to each group known; once read, the group is answered for real. + notReadyAfter = 30 * time.Millisecond + nb := newPatternBus() + nenv := readOnly.env() + nenv.Watch = nb.watch + nenv.Running = senv.Running + var nsaid []string + var nmu sync.Mutex + nenv.Say = func(l string) { nmu.Lock(); nsaid = append(nsaid, l); nmu.Unlock() } + nanswered := make(chan map[string]delivery.CandidateAnswer, 8) + nsh := NewShadow(nenv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { nanswered <- a }) + nctx, ncancel := context.WithCancel(ctx) + defer ncancel() + go func() { _ = nsh.Run(nctx) }() + nb.made(7) + nb.put(delivery.PlacementBucket, "placement.n1", tp, 1) + nb.put(delivery.HaveBucket, "n1.node", rec, 1) + for name, m := range mods { + raw, _ := delivery.EncodeModule(m) + nb.put(delivery.ModulesBucket, name, raw, 1) + } + sameRaw, _ := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "same", + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-same"}, Placement: &trunk}) + sameKey, _ := delivery.PlacementCandidateKey("same", "n1") + nb.put(delivery.PlacementCandidatesBucket, sameKey, sameRaw, 3) + for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { + nb.send(bucket, agent.Update{Caught: true}) + } + select { + case a := <-nanswered: + if g := a["same"]; g.Verdict != delivery.CandidateRefused || !strings.Contains(g.Refused, delivery.AssignerWantsBucket) || g.Head != "h-same" { + t.Errorf("a shadow waiting for its wants answered %+v", g) + } + case <-time.After(5 * time.Second): + t.Fatal("a shadow not ready after its wait answered nothing") + } + nmu.Lock() + if !strings.Contains(strings.Join(nsaid, "\n"), "this node answers no placement candidate yet: the node has not read "+delivery.AssignerWantsBucket) { + t.Errorf("a shadow not ready said %q", nsaid) + } + nmu.Unlock() + slowAfter = time.Nanosecond // every answer is slow: said, and noted in words that do not move with the time + defer func() { slowAfter = time.Minute }() + nb.send(delivery.AssignerWantsBucket, agent.Update{Caught: true}) + select { + case a := <-nanswered: + if g := a["same"]; g.Verdict != delivery.CandidateSame || !strings.Contains(g.Note, "this answer took over") { + t.Errorf("the shadow, once ready, answered %+v", g) + } + nmu.Lock() + if !strings.Contains(strings.Join(nsaid, "\n"), "the answer to same took") { + t.Errorf("a slow answer was not said: %q", nsaid) + } + nmu.Unlock() + case <-time.After(10 * time.Second): + t.Fatal("the shadow, once ready, answered nothing") + } + // The way back, row by row (review of the shadow, round 1): the controller's compositions at n1's wants compared // with what runs; one differing is answered so; one the controller holds none of refuses, without returning. var runsList []delivery.Composed diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index 4a8fea0..43818e3 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -1,6 +1,7 @@ package selfcompose import ( + "bytes" "context" "fmt" "sort" @@ -131,10 +132,19 @@ func (s *Shadow) Run(ctx context.Context) error { } caught := make([]bool, len(ins)) ready, dirty := false, false + // Not ready after a minute is said, naming each input not read yet, in the journal and in an answer to each group it + // 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() for { select { case <-ctx.Done(): return nil + case <-waiting.C: + if !ready { + s.notReady(ctx, ins, caught) + } + continue case t := <-merged: 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) @@ -229,6 +239,38 @@ func (s *Shadow) takeRecord(u agent.Update) bool { return had != has || (has && !delivery.SameNodeRecord(before, after)) } +// How long the shadow waits before it says it is not ready, and how long a group's answer runs before it is said slow. +var notReadyAfter, slowAfter = time.Minute, time.Minute + +// notReady says which inputs the shadow has not read yet, and answers each group it knows refused for it. +func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) { + var waits []string + for i, in := range ins { + if !caught[i] { + waits = append(waits, in.bucket+" ("+in.pattern+")") + } + } + why := fmt.Sprintf("the node has not read %s after %s; it answers once it has", strings.Join(waits, ", "), notReadyAfter) + s.c.env.Say("this node answers no placement candidate yet: " + why) + s.mu.Lock() + next := map[string]delivery.CandidateAnswer{} + for group, keys := range s.candidates { + a := delivery.CandidateAnswer{Refused: why, Verdict: delivery.CandidateRefused} + for n, k := range keys { + a.Head = k.c.Pull.Head + if n == s.c.env.Node { + a.Revision = k.rev + } + } + next[group] = a + } + s.answers = next + s.mu.Unlock() + if len(next) > 0 && s.answer != nil { + s.answer(ctx, next) + } +} + // answerAll answers every group and hands the answers when any changed. func (s *Shadow) answerAll(ctx context.Context) { c := s.c @@ -242,7 +284,20 @@ func (s *Shadow) answerAll(ctx context.Context) { sort.Strings(groups) next := map[string]delivery.CandidateAnswer{} for _, group := range groups { - if a, ok := s.answerGroup(ctx, group); ok { + began := time.Now() + slow := time.AfterFunc(slowAfter, func() { + c.env.Say(fmt.Sprintf("the answer to %s has run over %s, and goes on", group, slowAfter)) + }) + a, ok := s.answerGroup(ctx, group) + slow.Stop() + if took := time.Since(began); took > slowAfter { + c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second))) + if ok { + // Words that do not move with the time it took, so an answer the same as before is not written again. + a.Note = strings.TrimPrefix(a.Note+"; ", "; ") + "this answer took over " + slowAfter.String() + " to compose" + } + } + if ok { next[group] = a } } @@ -341,7 +396,8 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid c.mu.Lock() c.wants = held c.mu.Unlock() - a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn}, fps, running, head, rev) + a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn, + SameValue: c.sameValue}, fps, running, head, rev) return a, true } @@ -461,3 +517,17 @@ func returning(ctx context.Context, get func(ctx context.Context, bucket, key st a.Verdict = delivery.VerdictOf(a) return a } + +// sameValue says whether two sealed values open, with this node's key, to the same value; unknown when either does not +// open (or the node holds no opener). Only the comparison leaves: never a value (issue 1126, the shadow shows its work). +func (c *Composer) sameValue(a, b string) (same, known bool) { + if c.env.Unseal == nil { + return false, false + } + x, errA := c.env.Unseal(a) + y, errB := c.env.Unseal(b) + if errA != nil || errB != nil { + return false, false + } + return bytes.Equal(x, y), true +}