process: an awaited bucket says loudly once when the bus refuses hearing its creation, and hears it again and binds at each reconnect (issue 1790, review)

go/process/account_test.go

@@ -96,7 +96,7 @@ func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
 			t.Fatal("bound before the bucket exists")
 		}
 		bus.create()
-		hear.handle(&nats.Msg{Subject: hear.subject})
+		hear.deliver()
 		select {
 		case <-ready:
 		case <-time.After(5 * time.Second):
@@ -111,6 +111,31 @@ func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
 		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)
@@ -190,13 +215,45 @@ func (k *keptKV) Put(_ context.Context, key string, value []byte) (uint64, error
 	return uint64(len(k.puts)), nil
 }
 
-// heldSubscriber keeps the one subscription made, for the test to deliver to.
+// 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 {
-	subject string
+	mu         sync.Mutex
-	handle  nats.MsgHandler
+	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

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -87,12 +88,16 @@ type AwaitedBucket struct {
 	without string
 	logf    func(string, ...any)
 
-	mu      sync.Mutex
+	nc Subscriber
-	kv      jetstream.KeyValue
+
-	pending map[string][]byte
+	mu          sync.Mutex
-	order   []string
+	kv          jetstream.KeyValue
-	ready   []func(jetstream.KeyValue)
+	refusedSaid bool
-	sub     *nats.Subscription
+	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
@@ -105,12 +110,9 @@ func AwaitBucket(ctx context.Context, js BucketOpener, nc Subscriber, name, with
 	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() })
+		b.nc = nc
-		if err == nil {
+		b.hear()
-			b.mu.Lock()
+		b.followReconnects()
-			b.sub = sub
-			b.mu.Unlock()
-		}
 	}
 	ctxB, cancel := context.WithTimeout(ctx, bindWithin)
 	kv, err := js.KeyValue(ctxB, name)
@@ -204,12 +206,15 @@ func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.Ke
 		written++
 	}
 	b.kv, b.pending, b.order = kv, nil, nil
-	ready, sub := b.ready, b.sub
+	ready, sub, stop := b.ready, b.sub, b.stopStatus
-	b.ready, b.sub = nil, nil
+	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)
 	}
@@ -218,3 +223,100 @@ func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.Ke
 	}
 	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()
+			}
+		}
+	}()
+}