From d56c576e9ae97c82876c5c2907fb1ca72a5f93a1 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 02:44:39 +0100 Subject: [PATCH] Begin the shadow composer again with a backoff when an input cannot be watched yet, and pin mesh-sdk's fixed head, so a node answers candidates for the whole link (issue 1126, review of L2b round 1) --- cmd/mesh-host/nodeagent.go | 6 +---- go.mod | 2 +- go.sum | 2 ++ internal/selfcompose/selfcompose_test.go | 21 +++++++++++++++- internal/selfcompose/shadow.go | 32 ++++++++++++++++++++++++ 5 files changed, 56 insertions(+), 7 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 879d80c..8296b75 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -543,11 +543,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess func(ctx context.Context, answers map[string]delivery.CandidateAnswer) { session.NoteCandidates(ctx, answers) }) - go func() { - if err := shadow.Run(linked); err != nil && linked.Err() == nil { - say("this node answers no placement candidate on this link: " + err.Error()) - } - }() + go shadow.Keep(linked, link.Backoff) if composer := selfComposer(linked, session, node, say, hooks); composer != nil { env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy } else { diff --git a/go.mod b/go.mod index 5d3caf3..0a71ab3 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.20261027012704-062c29c3a22f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f 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 c719ba9..04ad7e0 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oa git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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 913335a..fe2dd79 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -355,7 +355,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a }) sctx, scancel := context.WithCancel(ctx) defer scancel() - go func() { _ = sh.Run(sctx) }() + // Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the + // indexer made it): the shadow tries again, and answers once it appears. + sb.missing(delivery.PlacementCandidatesBucket, true) + go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }) + time.Sleep(30 * time.Millisecond) + sb.missing(delivery.PlacementCandidatesBucket, false) sb.made(6) rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) tp, _ := delivery.EncodePlacement(trunk) @@ -433,6 +438,17 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { type patternBus struct { mu sync.Mutex watches []patternWatch + absent map[string]bool +} + +// missing makes a bucket's watch fail, as one the bus holds no state for, or work again. +func (b *patternBus) missing(bucket string, gone bool) { + b.mu.Lock() + defer b.mu.Unlock() + if b.absent == nil { + b.absent = map[string]bool{} + } + b.absent[bucket] = gone } type patternWatch struct { @@ -445,6 +461,9 @@ func newPatternBus() *patternBus { return &patternBus{} } func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) { b.mu.Lock() defer b.mu.Unlock() + if b.absent[bucket] { + return nil, fmt.Errorf("nats: bucket not found") + } w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)} b.watches = append(b.watches, w) return w.ch, nil diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index 20b4ce3..cef381c 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -5,6 +5,7 @@ import ( "fmt" "sort" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -44,6 +45,37 @@ func NewShadow(env Env, answer func(ctx context.Context, answers map[string]deli answers: map[string]delivery.CandidateAnswer{}} } +// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch +// yet (the candidates' state not made, or the node's grant of it not loaded at a delivery) or a watch that ended is +// tried again rather than leaving the node answering nothing for the rest of the link (review of L2b, round 1, finding +// 2). Each distinct reason is said once. +func (s *Shadow) Keep(ctx context.Context, backoff func(try int) time.Duration) { + said := "" + for try := 1; ctx.Err() == nil; try++ { + began := time.Now() + err := s.Run(ctx) + if ctx.Err() != nil { + return + } + if time.Since(began) > time.Minute { + try = 1 // it ran: a later stop begins the waits again + } + why := "it stopped" + if err != nil { + why = err.Error() + } + if why != said { + s.c.env.Say("this node answers no placement candidate for now, and tries again: " + why) + said = why + } + select { + case <-ctx.Done(): + return + case <-time.After(backoff(try)): + } + } +} + // Run answers until ctx ends or an input's watch ends. func (s *Shadow) Run(ctx context.Context) error { c := s.c