process: a bucket missing at a part's start is bound when the bus says it was created, keeping what was written meanwhile; buscompose grants each principal the creation of the buckets it may bind (issue 1790)

Since the grants part creates a manifest's buckets after a delivery (ADR 0547), a module restarted by that delivery can open a new state before it exists; OptionalBucket answered nil and the part never looked again (mesh-delivery's checker and builder, state hearing, 2026-10-26). AwaitBucket binds on the server's STREAM.CREATED advisory, or at the next use, and writes the latest value kept per key. The copied composer's agreement test leaves that one row out: the controller's list is no longer loaded since the take-over.
2026-10-26 15:11:00 +01:00
parent 21252194d5
commit d515c71e1c
10 changed files with 375 additions and 7 deletions
@@ -8,8 +8,11 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
+	"time"

+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )

@@ -73,6 +76,50 @@ func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
 		}
 	})

+	t.Run("1790: a bucket missing at start is bound when the bus says it was created, and what was written meanwhile is kept", func(t *testing.T) {
+		bus := &laterBus{}
+		hear := &heldSubscriber{}
+		var said []string
+		logf := func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
+		b := AwaitBucket(context.Background(), bus, hear, "m_hearing", "the health tool cannot say it", logf)
+		if hear.subject != "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_m_hearing" {
+			t.Fatalf("heard %q", hear.subject)
+		}
+		ready := make(chan jetstream.KeyValue, 1)
+		b.Ready(func(kv jetstream.KeyValue) { ready <- kv })
+		for _, v := range []string{"one", "two"} {
+			if _, err := b.Put(context.Background(), "dead-letters", []byte(v)); !errors.Is(err, ErrNotYet) {
+				t.Fatalf("a write while missing: %v", err)
+			}
+		}
+		if b.KV() != nil || len(ready) > 0 {
+			t.Fatal("bound before the bucket exists")
+		}
+		bus.create()
+		hear.handle(&nats.Msg{Subject: hear.subject})
+		select {
+		case <-ready:
+		case <-time.After(5 * time.Second):
+			t.Fatal("not bound within 5s of the bus saying the bucket was created")
+		}
+		if got := bus.kv.puts; strings.Join(got, ",") != "dead-letters=two" {
+			t.Errorf("written when it appeared: %v, want the latest value only", got)
+		}
+		if _, err := b.Put(context.Background(), "dead-letters", []byte("three")); err != nil {
+			t.Errorf("a write once bound: %v", err)
+		}
+		if len(said) != 2 || !strings.Contains(said[0], "cannot be read yet") || !strings.Contains(said[1], "appeared") {
+			t.Errorf("said %q", said)
+		}
+		// Without the advisory heard, the next use binds it.
+		quiet := &laterBus{}
+		c := AwaitBucket(context.Background(), quiet, nil, "m_hearing", "x", nil)
+		quiet.create()
+		if c.KV() == nil {
+			t.Error("a use after the bucket appeared did not bind it")
+		}
+	})
+
 	for _, tc := range []struct {
 		field  string
 		values map[string]any
@@ -114,3 +161,42 @@ func writeAccount(t *testing.T, fields map[string]any) string {
 type refusingBus struct{ err error }

 func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
+
+// laterBus has no bucket until create, and then one.
+type laterBus struct {
+	mu sync.Mutex
+	kv *keptKV
+}
+
+func (l *laterBus) create() { l.mu.Lock(); l.kv = &keptKV{}; l.mu.Unlock() }
+
+func (l *laterBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.kv == nil {
+		return nil, jetstream.ErrBucketNotFound
+	}
+	return l.kv, nil
+}
+
+// keptKV records each put as key=value.
+type keptKV struct {
+	jetstream.KeyValue
+	puts []string
+}
+
+func (k *keptKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
+	k.puts = append(k.puts, key+"="+string(value))
+	return uint64(len(k.puts)), nil
+}
+
+// heldSubscriber keeps the one subscription made, for the test to deliver to.
+type heldSubscriber struct {
+	subject string
+	handle  nats.MsgHandler
+}
+
+func (h *heldSubscriber) Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error) {
+	h.subject, h.handle = subject, handle
+	return nil, nil
+}
@@ -2,8 +2,12 @@ package process

 import (
 	"context"
+	"errors"
 	"fmt"
+	"sync"
+	"time"

+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )

@@ -53,3 +57,164 @@ func OptionalBucket(ctx context.Context, js BucketOpener, name, without string,
 func NotWatched(logf func(string, ...any), name string, err error, without string) {
 	logf("the bucket %s cannot be watched (%v): %s", name, err, without)
 }
+
+// ErrNotYet is a write to an awaited bucket that is not on the bus yet: the value is kept, the latest per key, and
+// written once the bucket appears.
+var ErrNotYet = errors.New("the bucket is not on the bus yet: the value is kept and written when it appears")
+
+// Subscriber is the part of a connection that hears what the bus says: a *nats.Conn, or a test's fake.
+type Subscriber interface {
+	Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error)
+}
+
+// createdAdvisory is the subject the bus says a bucket's stream was created on: the server's own advisory, which a
+// principal is granted for each bucket it may bind (mesh-sdk buscompose StreamCreatedAdvisory, stateCreatedHeard).
+func createdAdvisory(bucket string) string { return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket }
+
+// bindWithin bounds one bind of an awaited bucket: one STREAM.INFO.
+const bindWithin = 10 * time.Second
+
+// AwaitedBucket is a bucket a part keeps or reads that may not be on the bus at its start, bound once it is (novox/hq
+// issue 1790, met 2026-10-26: since the grants part creates a manifest's buckets after a delivery, ADR 0547, a module
+// restarted by that delivery opened its new state `hearing` before it existed, got "bucket not found", and never looked
+// again). It binds at AwaitBucket when it can; otherwise when the bus says the bucket's stream was created (heard on
+// nc), and at each use before then (one STREAM.INFO, the cheap check), never on a timer. It says once that it waits and
+// once that the bucket appeared.
+type AwaitedBucket struct {
+	ctx     context.Context
+	js      BucketOpener
+	name    string
+	without string
+	logf    func(string, ...any)
+
+	mu      sync.Mutex
+	kv      jetstream.KeyValue
+	pending map[string][]byte
+	order   []string
+	ready   []func(jetstream.KeyValue)
+	sub     *nats.Subscription
+}
+
+// AwaitBucket opens the bucket name now if the bus holds it, and otherwise waits for it as AwaitedBucket says; without
+// is what is not done while it is missing. nc may be nil: then only a use binds it. ctx is the part's life: a bind
+// after it ends is not tried.
+func AwaitBucket(ctx context.Context, js BucketOpener, nc Subscriber, name, without string, logf func(string, ...any)) *AwaitedBucket {
+	if logf == nil {
+		logf = func(string, ...any) {}
+	}
+	b := &AwaitedBucket{ctx: ctx, js: js, name: name, without: without, logf: logf, pending: map[string][]byte{}}
+	// Heard before the first try, so a creation between the try and the subscription is not missed.
+	if nc != nil {
+		sub, err := nc.Subscribe(createdAdvisory(name), func(*nats.Msg) { go b.bind() })
+		if err == nil {
+			b.mu.Lock()
+			b.sub = sub
+			b.mu.Unlock()
+		}
+	}
+	ctxB, cancel := context.WithTimeout(ctx, bindWithin)
+	kv, err := js.KeyValue(ctxB, name)
+	cancel()
+	if err == nil {
+		b.bound(kv, false)
+		return b
+	}
+	logf("the bucket %s cannot be read yet (%v): %s until it appears; it is bound when the bus says it was created, or at "+
+		"its next use", name, err, without)
+	return b
+}
+
+// KV is the bound bucket, or nil while it is not on the bus; while nil, the call tries to bind it once.
+func (b *AwaitedBucket) KV() jetstream.KeyValue {
+	b.mu.Lock()
+	kv := b.kv
+	b.mu.Unlock()
+	if kv != nil {
+		return kv
+	}
+	return b.bind()
+}
+
+// Ready runs f with the bucket: now when it is bound, otherwise once it appears (on the goroutine that bound it).
+func (b *AwaitedBucket) Ready(f func(jetstream.KeyValue)) {
+	b.mu.Lock()
+	kv := b.kv
+	if kv == nil {
+		b.ready = append(b.ready, f)
+	}
+	b.mu.Unlock()
+	if kv != nil {
+		f(kv)
+	}
+}
+
+// Put writes key, as jetstream.KeyValue's Put. While the bucket is missing it keeps value as key's latest and answers
+// ErrNotYet; the kept values are written, in the order their keys were first kept, when it appears.
+func (b *AwaitedBucket) Put(ctx context.Context, key string, value []byte) (uint64, error) {
+	kv := b.KV()
+	if kv == nil {
+		b.mu.Lock()
+		if b.kv == nil {
+			if _, ok := b.pending[key]; !ok {
+				b.order = append(b.order, key)
+			}
+			b.pending[key] = append([]byte(nil), value...)
+			b.mu.Unlock()
+			return 0, fmt.Errorf("%s: %w", b.name, ErrNotYet)
+		}
+		kv = b.kv
+		b.mu.Unlock()
+	}
+	return kv.Put(ctx, key, value)
+}
+
+// bind tries one STREAM.INFO of the bucket and, when it is there, binds it; it answers the bucket or nil.
+func (b *AwaitedBucket) bind() jetstream.KeyValue {
+	if b.ctx.Err() != nil {
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
+	defer cancel()
+	kv, err := b.js.KeyValue(ctx, b.name)
+	if err != nil {
+		return nil
+	}
+	return b.bound(kv, true)
+}
+
+// bound keeps kv as the bucket (unless another bind did first), writes what was kept and runs what waited on it.
+func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.KeyValue {
+	b.mu.Lock()
+	if b.kv != nil {
+		kv = b.kv
+		b.mu.Unlock()
+		return kv
+	}
+	// The kept values are written before the bucket is handed to anyone, so no later write is overtaken by an earlier.
+	written, failed := 0, 0
+	for _, key := range b.order {
+		ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
+		_, err := kv.Put(ctx, key, b.pending[key])
+		cancel()
+		if err != nil {
+			failed++
+			b.logf("the bucket %s appeared but the value kept for %s could not be written: %v", b.name, key, err)
+			continue
+		}
+		written++
+	}
+	b.kv, b.pending, b.order = kv, nil, nil
+	ready, sub := b.ready, b.sub
+	b.ready, b.sub = nil, nil
+	b.mu.Unlock()
+	if sub != nil {
+		_ = sub.Unsubscribe()
+	}
+	if appeared {
+		b.logf("the bucket %s appeared and is bound: %d kept value(s) written, %d not", b.name, written, failed)
+	}
+	for _, f := range ready {
+		f(kv)
+	}
+	return kv
+}
@@ -58,7 +58,9 @@ func grantedSendProcess(t *testing.T) (url string, account Account) {
 			{Username: adminUser, Password: adminPass},
 			{Username: sendUser, Password: sendPassword, Permissions: &server.Permissions{
 				Publish:   &server.SubjectPermission{Allow: sendGrants},
-				Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>"}},
+				// And the creation of its bucket, as mesh-sdk's composer grants it since issue 1790 (stateCreatedHeard).
+				Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>",
+					"$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-mail_sends"}},
 			}},
 		},
 	}
@@ -18,12 +18,40 @@ import (
 // key, reads it, and updates it with the revision it read, then each succeeds and returns the new
 // revision; an update with a stale revision fails as stdio.ErrChanged; a create of an existing key fails
 // as stdio.ErrChanged.
+//
+// 1790: and before that, the bucket is not on the bus when the process opens it (a delivery restarted it before the
+// grants part created its new state): it is bound when the bus says the bucket was created, as the process's own
+// account hears it, and the value written while it was missing is there.
 func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) {
 	url, account := grantedSendProcess(t)
 	_, adminJetStream := adminJS(t, url)
+
+	nc, err := nats.Connect(url, nats.UserInfo(sendUser, sendPassword), nats.CustomInboxPrefix("_INBOX."+sendUser))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	awaited := AwaitBucket(ctxFor(t), js, nc, "mesh-mail_sends", "nothing is kept", t.Logf)
+	appeared := make(chan struct{})
+	awaited.Ready(func(jetstream.KeyValue) { close(appeared) })
+	if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":false}`)); !errors.Is(err, ErrNotYet) {
+		t.Fatalf("a write before the bucket exists: %v, want ErrNotYet", err)
+	}
+	if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":true}`)); !errors.Is(err, ErrNotYet) {
+		t.Fatalf("a second write before the bucket exists: %v, want ErrNotYet", err)
+	}
 	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
 		t.Fatalf("the admin could not make the bucket: %v", err)
 	}
+	select {
+	case <-appeared:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the bucket was not bound within 5s of its creation, with no use since")
+	}

 	conn, err := Connect(account)
 	if err != nil {
@@ -31,6 +59,9 @@ func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) {
 	}
 	defer conn.Close()
 	s := conn.State("sends")
+	if kept, err := s.Get("hearing"); err != nil || kept == nil || string(kept.Value) != `{"heard":true}` {
+		t.Fatalf("the value written while the bucket was missing: %+v, %v; want the latest", kept, err)
+	}

 	rev, err := s.Create("key", map[string]any{"v": 1})
 	if err != nil {