Node agent: wait for the bus's word that the seat's placement and candidates states were created, then watch them (issue 1126)

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")
+		}
+	}
 }

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
+}