diff --git a/go/process/account_test.go b/go/process/account_test.go index 8b60dd8..7c7002b 100644 --- a/go/process/account_test.go +++ b/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 - handle nats.MsgHandler + 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}) +} diff --git a/go/process/buckets.go b/go/process/buckets.go index c706ba1..bc2421c 100644 --- a/go/process/buckets.go +++ b/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 - kv jetstream.KeyValue - pending map[string][]byte - order []string - ready []func(jetstream.KeyValue) - sub *nats.Subscription + 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 @@ -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() }) - if err == nil { - b.mu.Lock() - b.sub = sub - b.mu.Unlock() - } + b.nc = nc + b.hear() + b.followReconnects() } 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 - b.ready, b.sub = 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) } @@ -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() + } + } + }() +}