Merge pull request 'A module waits for a state created after it starts (issue 1790)' (#97) from fix/1790-state-appears into main

go/buscompose/buscompose_test.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"regexp"
 	"strings"
 	"testing"
 )
@@ -14,6 +15,9 @@ import (
 // internal/broker (mesh-controller branch fix/1945-mesh-grants-publishes-memberships, and fix/1790-grants-create-buckets for the grants part creating declared streams), lists and memberships, from fixtures_test.go compiled there unchanged; composed.conf is the
 // controller's golden of the whole server configuration. A row that differs is the copy drifting from the controller,
 // which is a fault while both compose (until the controller loses its lease, ADR 0547 §13).
+// createdAdvisory is one subscription to a bucket's creation in a composed list, as stateCreatedHeard composes it.
+var createdAdvisory = regexp.MustCompile(`"\$JS\.EVENT\.ADVISORY\.STREAM\.CREATED\.KV_[^"]*", `)
+
 func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
 	raw, err := os.ReadFile(filepath.Join("testdata", "controller-user-lists.json"))
 	if err != nil {
@@ -48,6 +52,11 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
 			if strings.Join(missing, ",") != strings.Join(want[i].Missing, ",") {
 				t.Errorf("left out %v, the controller %v", missing, want[i].Missing)
 			}
+			// The one row the copy composes beyond the controller's, on purpose (novox/hq issue 1790, 2026-10-26, after
+			// the take-over of ADR 0547 milestone 2): each principal hears its own and read buckets created, so a state
+			// missing at its start is bound when it appears. The controller's composer is frozen and its list no longer
+			// loaded; while it is not, this row is left out of the comparison and nothing else.
+			list = createdAdvisory.ReplaceAllString(list, "")
 			if list != want[i].List {
 				t.Errorf("the list differs from the controller's:\n%s", list)
 			}

go/buscompose/creates_streams_test.go

@@ -90,6 +90,22 @@ func TestTheGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
 			t.Errorf("%s may publish %s: %v, want %v", row.who, row.subject, got, row.may)
 		}
 	}
+	// Issue 1790 (2026-10-26): a module hears the bus create each bucket it may bind, its own and those it reads, so a
+	// state missing at its start is bound when it appears; no other bucket's, and no other advisory.
+	for _, row := range []struct {
+		who, subject string
+		may          bool
+	}{
+		{"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-delivery-builder_steps", true},
+		{GrantsModule, "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-vault_bus-accounts", true},
+		{"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-grants_assertions", false},
+		{"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.CREATED.LOG_mesh-delivery-builder_builds", false},
+		{"mesh-delivery-builder", "$JS.EVENT.ADVISORY.STREAM.DELETED.KV_mesh-delivery-builder_steps", false},
+	} {
+		if got := MaySubscribe(perms[row.who], row.subject); got != row.may {
+			t.Errorf("%s may hear %s: %v, want %v", row.who, row.subject, got, row.may)
+		}
+	}
 	// A name that is no plain token is never granted, and a module other than the grants part is never granted one.
 	if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 2 {
 		t.Errorf("grants for wildcard names: %v", got)

go/buscompose/nats.go

@@ -1085,8 +1085,12 @@ func PermissionsFor(p Principal) (Permissions, error) {
 
 		// 5. Its state, and the state of others it reads (novox/hq ADR 0201): every one read and
 		//    watched, its own written too.
-		pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State,
+		kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State, PerMachine: p.PerMachine, Reads: p.Reads,
-			PerMachine: p.PerMachine, Reads: p.Reads, KeyedReads: p.KeyedReads})...)
+			KeyedReads: p.KeyedReads}
+		pub = append(pub, stateGrants(kept)...)
+		// And it hears each of those buckets created, so one not on the bus at its start is bound when it appears
+		// (novox/hq issue 1790; mesh-sdk process.AwaitBucket).
+		sub = append(sub, stateCreatedHeard(kept)...)
 		// And its logs (novox/hq ADR 0297): appended to and read directly, its own alone.
 		pub = append(pub, logGrants(p.Module, p.Logs)...)
 

go/buscompose/process_grants.go

@@ -30,7 +30,11 @@ func processGrants(p Principal) (pub, sub []string, err error) {
 
 	// 3. Its module's own state, by the local names its bus names alone — never the whole of what the
 	// module keeps (novox/hq ADR 0436 §1), read and written as stateGrants gives any module's own bucket.
-	pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State})...)
+	kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State}
+	pub = append(pub, stateGrants(kept)...)
+	// And it hears each of them created, so one not on the bus at its start is bound when it appears (novox/hq issue
+	// 1790; process.AwaitBucket).
+	sub = append(sub, stateCreatedHeard(kept)...)
 
 	// 4. The router's records of its module's own asks (novox/hq ADR 0436 §1, ADR 0259 §3): the same
 	// direct-get subjects a module's own principal reads for the seats its Uses names that keep them —

go/buscompose/state.go

@@ -166,6 +166,49 @@ func stateGrants(a stateAccess) []string {
 	return out
 }
 
+// StreamCreatedAdvisory is the subject the bus says a stream was created on (the server's own advisory), here for a
+// bucket's stream KV_<bucket>.
+func StreamCreatedAdvisory(bucket string) string {
+	return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket
+}
+
+// stateCreatedHeard is what a principal subscribes to so that a bucket it may bind and that is not on the bus yet is
+// bound when it appears (novox/hq issue 1790, 2026-10-26: since the grants part creates a manifest's buckets, ADR 0547,
+// a module's delivery can start it before its new state exists): the bus's advisory that the bucket's stream was
+// created, for each bucket stateGrants lets it bind, and no other. Read-only: the server publishes it in the mesh's own
+// account, and hearing it changes nothing.
+func stateCreatedHeard(a stateAccess) []string {
+	var out []string
+	seen := map[string]bool{}
+	add := func(bucket string) {
+		if !seen[bucket] {
+			seen[bucket] = true
+			out = append(out, StreamCreatedAdvisory(bucket))
+		}
+	}
+	perMachine := map[string]bool{}
+	for _, n := range a.PerMachine {
+		perMachine[n] = true
+	}
+	for _, name := range a.Keeps {
+		if !safeSubject.MatchString(name) || (perMachine[name] && !safeSubject.MatchString(a.Node)) {
+			continue
+		}
+		add(BucketName(a.Module, name))
+	}
+	for _, r := range a.Reads {
+		if bucket, ok := bucketOfRead(r); ok {
+			add(bucket)
+		}
+	}
+	for _, k := range a.KeyedReads {
+		if bucket, ok := bucketOfRead(k.Read); ok && safeSubject.MatchString(k.Key) {
+			add(bucket)
+		}
+	}
+	return out
+}
+
 // KeyedRead is one key of another module's state a principal reads, and no other key of it.
 type KeyedRead struct {
 	Read string `json:"read"`

go/delivery/buildrun/lostask.go

@@ -3,6 +3,7 @@ package buildrun
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -13,6 +14,7 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/health"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask
@@ -458,9 +460,15 @@ func (h *DeadLetterHearing) tell() {
 // health tool reads.
 const HearingState = "hearing"
 
+// HearingKeeper is where a Hearing is kept: a jetstream.KeyValue, or a process.AwaitedBucket for a state that may not be
+// on the bus at the part's start (issue 1790), which keeps the latest until it appears and answers process.ErrNotYet.
+type HearingKeeper interface {
+	Put(ctx context.Context, key string, value []byte) (uint64, error)
+}
+
 // SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and
-// says so once.
+// says so once. A write kv keeps for later (process.ErrNotYet) is not an error: the awaited bucket said it waits.
-func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Hearing) {
+func SayHearing(ctx context.Context, kv HearingKeeper, logf func(string, ...any)) func(Hearing) {
 	if kv == nil {
 		if logf != nil {
 			logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState)
@@ -472,7 +480,7 @@ func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ..
 		if err == nil {
 			_, err = kv.Put(ctx, HearingKey, data)
 		}
-		if err != nil && logf != nil {
+		if err != nil && !errors.Is(err, process.ErrNotYet) && logf != nil {
 			logf("whether %s is heard could not be kept: %v", DeadLettersStream, err)
 		}
 	}

go/process/account_test.go

@@ -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,75 @@ 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.deliver()
+		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)
+		}
+		// The bus refuses the subscription to the creation: said once, loudly; a reconnect hears it again and binds.
+		refusedBus, refusing := &laterBus{}, &heldSubscriber{refused: true}
+		var loud []string
+		r := AwaitBucket(context.Background(), refusedBus, refusing, "m_reasks", "x",
+			func(f string, a ...any) { loud = append(loud, fmt.Sprintf(f, a...)) })
+		bound := make(chan struct{})
+		r.Ready(func(jetstream.KeyValue) { close(bound) })
+		if len(loud) != 2 || !strings.Contains(loud[0], "cannot hear the creation of m_reasks") ||
+			!strings.Contains(loud[0], "permissions violation") {
+			t.Fatalf("a refused subscription was said as %q", loud)
+		}
+		refusing.mu.Lock()
+		refusing.refused = false
+		refusing.mu.Unlock()
+		refusedBus.create()
+		refusing.status <- nats.CONNECTED
+		select {
+		case <-bound:
+		case <-time.After(5 * time.Second):
+			t.Fatal("not bound after the connection connected again")
+		}
+		if refusing.subscribed != 2 {
+			t.Errorf("subscribed %d times, want again at the reconnect", refusing.subscribed)
+		}
+
+		// 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 +186,74 @@ 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 last subscription made, for the test to deliver to; refused, the bus refuses it as a
+// *nats.Conn says it (the connection's last error after a round trip). status stands for the connection's CONNECTED.
+type heldSubscriber struct {
+	mu         sync.Mutex
+	subject    string
+	handle     nats.MsgHandler
+	subscribed int
+	refused    bool
+	last       error
+	status     chan nats.Status
+}
+
+func (h *heldSubscriber) Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.subject, h.handle = subject, handle
+	h.subscribed++
+	if h.refused {
+		h.last = fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject)
+	}
+	return nil, nil
+}
+
+func (h *heldSubscriber) FlushTimeout(time.Duration) error { return nil }
+
+func (h *heldSubscriber) LastError() error { h.mu.Lock(); defer h.mu.Unlock(); return h.last }
+
+func (h *heldSubscriber) StatusChanged(...nats.Status) chan nats.Status {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.status = make(chan nats.Status, 1)
+	return h.status
+}
+
+func (h *heldSubscriber) RemoveStatusListener(chan nats.Status) {}
+
+func (h *heldSubscriber) deliver() {
+	h.mu.Lock()
+	handle, subject := h.handle, h.subject
+	h.mu.Unlock()
+	handle(&nats.Msg{Subject: subject})
+}

go/process/buckets.go

@@ -2,8 +2,13 @@ package process
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"strings"
+	"sync"
+	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
@@ -53,3 +58,265 @@ 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)
+
+	nc Subscriber
+
+	mu          sync.Mutex
+	kv          jetstream.KeyValue
+	refusedSaid bool
+	stopStatus  func()
+	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 {
+		b.nc = nc
+		b.hear()
+		b.followReconnects()
+	}
+	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, stop := b.ready, b.sub, b.stopStatus
+	b.ready, b.sub, b.stopStatus = nil, nil, nil
+	b.mu.Unlock()
+	if sub != nil {
+		_ = sub.Unsubscribe()
+	}
+	if stop != nil {
+		stop()
+	}
+	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
+}
+
+// refusalReader is the part of a *nats.Conn that says whether the bus refused a subscription: the server's refusal comes
+// back after a round trip as the connection's last error, naming the subject.
+type refusalReader interface {
+	FlushTimeout(time.Duration) error
+	LastError() error
+}
+
+// statusFollower is the part of a *nats.Conn that says when it connected again, without taking its reconnect handler.
+type statusFollower interface {
+	StatusChanged(statuses ...nats.Status) chan nats.Status
+	RemoveStatusListener(ch chan nats.Status)
+}
+
+// hear subscribes to the bucket's creation (again), and says once, loudly, when the bus refuses it: then only a use or
+// a reconnect binds the bucket.
+func (b *AwaitedBucket) hear() {
+	subject := createdAdvisory(b.name)
+	b.mu.Lock()
+	old, bound := b.sub, b.kv != nil
+	b.sub = nil
+	b.mu.Unlock()
+	if old != nil {
+		_ = old.Unsubscribe()
+	}
+	if bound {
+		return
+	}
+	checks, _ := b.nc.(refusalReader)
+	var before error
+	if checks != nil {
+		before = checks.LastError()
+	}
+	sub, err := b.nc.Subscribe(subject, func(*nats.Msg) { go b.bind() })
+	if err == nil && checks != nil && checks.FlushTimeout(bindWithin) == nil {
+		if after := checks.LastError(); after != nil && after != before && errors.Is(after, nats.ErrPermissionViolation) &&
+			strings.Contains(after.Error(), subject) {
+			err = after
+		}
+	}
+	if err != nil {
+		if sub != nil {
+			_ = sub.Unsubscribe()
+		}
+		b.mu.Lock()
+		said := b.refusedSaid
+		b.refusedSaid = true
+		b.mu.Unlock()
+		if !said {
+			b.logf("cannot hear the creation of %s: %v; it binds at its next use, or when the connection reconnects", b.name, err)
+		}
+		return
+	}
+	b.mu.Lock()
+	if b.kv != nil {
+		b.mu.Unlock()
+		if sub != nil {
+			_ = sub.Unsubscribe()
+		}
+		return
+	}
+	b.sub, b.refusedSaid = sub, false
+	b.mu.Unlock()
+}
+
+// followReconnects hears the bucket's creation again and tries a bind each time the connection connects again (a grant
+// reaches the bus with a new user list; a lost subscription is made again), until the bucket is bound or the part ends.
+func (b *AwaitedBucket) followReconnects() {
+	w, ok := b.nc.(statusFollower)
+	if !ok {
+		return
+	}
+	ch := w.StatusChanged(nats.CONNECTED)
+	b.mu.Lock()
+	if b.kv != nil {
+		b.mu.Unlock()
+		w.RemoveStatusListener(ch)
+		return
+	}
+	b.stopStatus = func() { w.RemoveStatusListener(ch) }
+	b.mu.Unlock()
+	go func() {
+		for {
+			select {
+			case <-b.ctx.Done():
+				w.RemoveStatusListener(ch)
+				return
+			case _, open := <-ch:
+				if !open {
+					return
+				}
+				b.hear()
+				b.KV()
+			}
+		}
+	}()
+}

go/process/fixture_integration_test.go

@@ -57,8 +57,10 @@ func grantedSendProcess(t *testing.T) (url string, account Account) {
 		Users: []*server.User{
 			{Username: adminUser, Password: adminPass},
 			{Username: sendUser, Password: sendPassword, Permissions: &server.Permissions{
-				Publish:   &server.SubjectPermission{Allow: sendGrants},
+				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"}},
 			}},
 		},
 	}

go/process/state_integration_test.go

@@ -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 {