Merge pull request 'Node shadow: decides 'same value' by opening seals in memory, says when it is not ready or slow, carries the bus membership (issue 1126, group fix/1126-shadow-shows)' (#156) from fix/1126-shadow-shows into main

2026-10-28 00:21:51 +00:00
parent 449cda1b00 047237c911
commit cff69893a3
4 changed files with 134 additions and 3 deletions
@@ -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
@@ -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=
@@ -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
@@ -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
+}