The shadow shows its work: differing resources with masked bodies and whether a sealed value is the same, a minute not ready or a slow answer said in the journal and the answer; pin mesh-sdk 4e0c577 (issue 1126, seed #531)
mesh/delivery merged: source.main moved to novox/mesh-host@cff69893a361
mesh/delivery-group group fix/1126-shadow-shows merged: every member is merged
mesh/check pass 047237c911a4 on 449cda1b003e | unit 2980 run, 270 kept, 36 skipped | integration 2999 run, 302 kept, 29 skipped | over bound: unit (sl…
mesh/delivery merged: source.main moved to novox/mesh-host@cff69893a361
mesh/delivery-group group fix/1126-shadow-shows merged: every member is merged
mesh/check pass 047237c911a4 on 449cda1b003e | unit 2980 run, 270 kept, 36 skipped | integration 2999 run, 302 kept, 29 skipped | over bound: unit (sl…
ace, g14 and shanks answered nothing to seed #531 and said nothing either. The shadow now says, after a minute not ready, which inputs it has not read, and answers each group it knows refused for that; an answer running over a minute is said while it runs and noted in the answer. The answer's differing resources carry both bodies, masked, the node saying whether two sealed values open to the same value (Composer.sameValue: only the comparison leaves the node).
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user