From dd8ba97989bc69b7a3a71daf7ca4823eaeb1b50a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:43:48 +0100 Subject: [PATCH] Node agent: wait for the bus's word that the seat's placement and candidates states were created, then watch them (issue 1126) --- cmd/mesh-host/nodeagent.go | 38 ++++++++++++--- internal/link/history_integration_test.go | 33 +++++++++++++ internal/link/watch.go | 57 +++++++++++++++++++++++ 3 files changed, 122 insertions(+), 6 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 0a050bb..430be23 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -39,6 +39,20 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up if err != nil { return nil, err } + return w.updates(ctx, in), nil +} + +// WatchAwait is Watch for a state that its holder makes when it starts (the placement seat's): a bucket not on the bus +// yet is waited for on the bus's word that it was created (link.Session.KVWatchAwait), and waiting is told why. +func (w bucketWatch) WatchAwait(ctx context.Context, pattern string, waiting func(error)) (<-chan agent.Update, error) { + in, err := w.session.KVWatchAwait(ctx, w.bucket, pattern, waiting) + if err != nil { + return nil, err + } + return w.updates(ctx, in), nil +} + +func (w bucketWatch) updates(ctx context.Context, in <-chan link.KVUpdate) <-chan agent.Update { out := make(chan agent.Update, 64) go func() { defer close(out) @@ -50,7 +64,7 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up } } }() - return out, nil + return out } // sessionHaves writes the haves the agent says, through the session. @@ -926,6 +940,13 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { return selfcompose.Env{Node: node, Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { + if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket { + // The seat's states are made when their holder starts: a node-engine that starts first waits for the + // bus's word that the bucket was created, and the shadow and the composer say they wait. + return bucketWatch{session, bucket}.WatchAwait(ctx, pattern, func(err error) { + say("the seat's state " + bucket + " is not on the bus yet (" + err.Error() + "): it is watched when the bus says it was created") + }) + } return bucketWatch{session, bucket}.Watch(ctx, pattern) }, Get: func(ctx context.Context, bucket, key string) ([]byte, error) { @@ -988,11 +1009,16 @@ func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop if err != nil { return } - // Tried at the start only: the delivery brings the grants part and the holder first, and the holder makes its states - // at its own start, so the state exists and the read is granted by the time the node-engine starts. A watch still - // refused is said once in the journal and in the node's own have, where the watchman reads it, and the node stays - // on the controller's compositions until its next start. - updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key) + // The seat's placement state is made when its holder starts; the delivery brings the grants part and the holder first, + // so it is normally there. A node-engine that starts before it waits for the bus's word that the bucket was created + // (no timer), says so once in the journal and in the node's own have, where the watchman reads it, and stays on the + // controller's compositions until its key is written. A watch the account may not make is said the same way, and the + // node stays until its next start. + updates, err := bucketWatch{session, delivery.PlacementBucket}.WatchAwait(ctx, key, func(err error) { + why := "this node's key in the seat's placement state is waited for, the state is not on the bus yet (" + err.Error() + "): it stays on the controller's compositions and is watched when the bus says the state was created" + say(why) + note(why) + }) if err != nil { why := "this node's key in the seat's placement state cannot be followed (" + err.Error() + "): it stays on the controller's compositions until its next start" say(why) diff --git a/internal/link/history_integration_test.go b/internal/link/history_integration_test.go index 4f042b3..38e1617 100644 --- a/internal/link/history_integration_test.go +++ b/internal/link/history_integration_test.go @@ -4,6 +4,7 @@ package link import ( "context" + "sync/atomic" "testing" "time" @@ -90,4 +91,36 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) { } else { t.Logf("a refused watch is an error after %s: %v", time.Since(started).Round(time.Millisecond), err) } + + // A bucket made after the watch was asked is then watched, on the bus's word that it was created and with no timer; + // a bucket the account may not read is no waiting but the error (endWhenPlaced, the shadow's candidates). + q := &Session{Membership: Membership{Node: "n1"}} + q.Attach(ctx, OverNATS{Conn: conn, JS: js}) + var waited atomic.Int32 + actx, acancel := context.WithTimeout(ctx, 20*time.Second) + defer acancel() + ch, err := q.KVWatchAwait(actx, "later-it", ">", func(error) { waited.Add(1) }) + if err != nil || waited.Load() != 1 { + t.Fatalf("a bucket not there yet: watch %v, waiting told %d time(s)", err, waited.Load()) + } + select { + case u := <-ch: + t.Fatalf("an update before the bucket exists: %+v", u) + case <-time.After(200 * time.Millisecond): + } + later, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "later-it", History: 1}) + if err != nil { + t.Fatal(err) + } + if _, err := later.PutString("placement.n1", "v"); err != nil { + t.Fatal(err) + } + for seen := false; !seen; { + select { + case u := <-ch: + seen = u.Key == "placement.n1" && string(u.Value) == "v" + case <-actx.Done(): + t.Fatal("the bucket made after the watch was never watched") + } + } } diff --git a/internal/link/watch.go b/internal/link/watch.go index a1f5687..4d6eeb0 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -239,3 +239,60 @@ func (q *Session) Hear(subject string) (<-chan []byte, func(), error) { } return h.Hear(subject) } + +// CreatedAdvisory is the subject the bus says a bucket's stream was created on (the server's own advisory; the same +// subject mesh-sdk's process.AwaitBucket binds on, and the grants part gives a principal for each bucket it may read). +func CreatedAdvisory(bucket string) string { + return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket +} + +// KVWatchAwait is KVWatch for a bucket that may not be on the bus yet (the placement seat's states, made when their +// holder starts): the channel is returned at once, and while the bucket is missing it yields nothing and waiting is told +// why; when the bus says the bucket's stream was created the watch is made and its updates follow. No timer: the +// advisory is heard before the first try, so a creation between the try and the wait is not missed. Any other error of +// the first try is the caller's, as for KVWatch: a bucket the account may not read is not waited for. +func (q *Session) KVWatchAwait(ctx context.Context, bucket, pattern string, waiting func(error)) (<-chan KVUpdate, error) { + created, stop, err := q.Hear(CreatedAdvisory(bucket)) + if err != nil { + return nil, err + } + in, err := q.KVWatch(ctx, bucket, pattern) + if err == nil { + stop() + return in, nil + } + if !errors.Is(err, nats.ErrBucketNotFound) { + stop() + return nil, err + } + waiting(err) + out := make(chan KVUpdate, 64) + go func() { + defer close(out) + defer stop() + for in == nil { + select { + case <-ctx.Done(): + return + case _, ok := <-created: + if !ok { + return + } + w, err := q.KVWatch(ctx, bucket, pattern) + if err != nil { + waiting(err) + continue + } + in = w + } + } + for u := range in { + select { + case out <- u: + case <-ctx.Done(): + return + } + } + }() + return out, nil +}