Catch a denied publish at once instead of waiting it out, so an ungranted bucket's refusal names it (issue 1188, review round 1 finding 1)

go/process/asks.go

@@ -24,7 +24,7 @@ func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) {
 		return nil, err
 	}
 	subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id
-	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
 	defer cancel()
 	reply, err := c.nc.RequestWithContext(ctx, subject, nil)
 	if err != nil {

go/process/conn.go

@@ -1,12 +1,15 @@
 package process
 
 import (
+	"context"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"regexp"
+	"sync"
 
 	"github.com/nats-io/nats.go"
 )
@@ -17,13 +20,43 @@ import (
 type Conn struct {
 	nc      *nats.Conn
 	account Account
+
+	mu      sync.Mutex
+	next    uint64
+	pending map[string]map[uint64]context.CancelCauseFunc
 }
 
+// errPermissionRefused marks a context cancelled because the bus refused this account's publish of the
+// subject a pending call was waiting an answer on: a denied publish is refused only as an async protocol
+// error with no reply (mesh-tools node-tools/internal/bus/caller.go publishRefused), so without this a
+// call would wait out its whole timeout instead of failing at once.
+var errPermissionRefused = errors.New("the bus refused this account's publish")
+
+// refusedPublish reads the subject out of the bus's "-ERR 'Permissions Violation for Publish to
+// "<subject>"'" (mesh-tools node-tools/internal/bus/caller.go's own regexp).
+var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`)
+
 // Connect dials the bus with account: pinned to its fingerprint when it carries one (refused with any
 // other certificate), authenticated as its user and password, on the inbox prefix _INBOX.<user> and no
 // other. Ported from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned,
 // PinnedToFingerprint) since this package may not import the controller.
+//
+// An ErrorHandler among opts is chained behind this package's own, which watches every async error for a
+// permission violation refusing a publish this connection is waiting an answer to, and cancels that wait
+// at once.
 func Connect(account Account, opts ...nats.Option) (*Conn, error) {
+	c := &Conn{account: account, pending: map[string]map[uint64]context.CancelCauseFunc{}}
+
+	// opts are functions over a *nats.Options; applied to a scratch value, this reads off a caller's own
+	// ErrorHandler (if any) to chain, without needing nats.Connect's own URL/Servers parsing here.
+	var scratch nats.Options
+	for _, opt := range opts {
+		if opt != nil {
+			_ = opt(&scratch)
+		}
+	}
+	chained := scratch.AsyncErrorCB
+
 	all := []nats.Option{
 		nats.CustomInboxPrefix(account.InboxPrefix()),
 		nats.UserInfo(account.User, account.Password),
@@ -32,16 +65,72 @@ func Connect(account Account, opts ...nats.Option) (*Conn, error) {
 		all = append(all, nats.Secure(pinnedToFingerprint(account.Fingerprint)))
 	}
 	all = append(all, opts...)
+	// Last, so it is the one AsyncErrorCB nats.Connect keeps however the opts above ordered theirs.
+	all = append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
+		c.publishRefused(err)
+		if chained != nil {
+			chained(nc, sub, err)
+		}
+	}))
 	nc, err := nats.Connect(account.URL, all...)
 	if err != nil {
 		return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err)
 	}
-	return &Conn{nc: nc, account: account}, nil
+	c.nc = nc
+	return c, nil
 }
 
 // Close ends the connection.
 func (c *Conn) Close() { c.nc.Close() }
 
+// withRefusal derives a context from ctx that is also cancelled, with errPermissionRefused as its cause,
+// the moment the bus refuses this account's publish of subject — before ctx's own deadline, which a
+// denied publish would otherwise run out, refused only as an async error with no reply. Every call that
+// requests on the bus registers the subject it is about to publish to here first.
+func (c *Conn) withRefusal(ctx context.Context, subject string) (context.Context, func()) {
+	derived, cancel := context.WithCancelCause(ctx)
+	c.mu.Lock()
+	c.next++
+	id := c.next
+	if c.pending[subject] == nil {
+		c.pending[subject] = map[uint64]context.CancelCauseFunc{}
+	}
+	c.pending[subject][id] = cancel
+	c.mu.Unlock()
+	return derived, func() {
+		c.mu.Lock()
+		if list := c.pending[subject]; list != nil {
+			delete(list, id)
+			if len(list) == 0 {
+				delete(c.pending, subject)
+			}
+		}
+		c.mu.Unlock()
+		cancel(nil)
+	}
+}
+
+// publishRefused ends every call waiting on the subject the bus's error names, with errPermissionRefused
+// as the cause (mesh-tools node-tools/internal/bus/caller.go publishRefused): true when there was one
+// waiting.
+func (c *Conn) publishRefused(err error) bool {
+	if !errors.Is(err, nats.ErrPermissionViolation) {
+		return false
+	}
+	m := refusedPublish.FindStringSubmatch(err.Error())
+	if len(m) < 2 {
+		return false
+	}
+	c.mu.Lock()
+	cancels := c.pending[m[1]]
+	delete(c.pending, m[1])
+	c.mu.Unlock()
+	for _, cancel := range cancels {
+		cancel(errPermissionRefused)
+	}
+	return len(cancels) > 0
+}
+
 // pinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its
 // name and chain deliberately not consulted, as mesh-controller's own pin does (jetstream.go:116-132).
 func pinnedToFingerprint(want string) *tls.Config {

go/process/emit.go

@@ -51,7 +51,7 @@ func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
 	defer cancel()
 	_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
 	return err

go/process/state.go

@@ -7,16 +7,16 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 )
 
-// StateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a var, not a const, so a
-// test whose account is not granted a subject — refused only as an async protocol error with no reply,
-// never a reply naming the refusal — need not wait out the full bound (mesh-tools' own ProofTimeout,
-// node-tools/internal/bus/seat.go:65, is a var for the same reason).
-var StateTimeout = 30 * time.Second
+// stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is
+// caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so
+// this stays the ordinary request bound and is not exposed for a test to shrink.
+const stateTimeout = 30 * time.Second
 
 // State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436
 // clause 4), in the bucket <module>_<name> (mesh-controller internal/broker/state.go:49) — only as the
@@ -33,14 +33,27 @@ type State struct {
 func (s State) bucket() string { return s.Module + "_" + s.Name }
 
 // State is the named state this connection's module keeps, bound lazily to its bucket on each call so a
-// bucket the account is not granted fails at the call that needed it, naming the bucket.
+// bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself
+// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
+// not granted is refused there, promptly, as nats.ErrPermissionViolation.
 func (c *Conn) State(name string) State {
+	bucket := c.account.Module + "_" + name
 	return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
 		js, err := jetstream.New(c.nc)
 		if err != nil {
 			return nil, err
 		}
-		return js.KeyValue(ctx, c.account.Module+"_"+name)
+		subject := "$JS.API.STREAM.INFO.KV_" + bucket
+		ctx, stop := c.withRefusal(ctx, subject)
+		defer stop()
+		kv, err := js.KeyValue(ctx, bucket)
+		if err != nil {
+			if errors.Is(context.Cause(ctx), errPermissionRefused) {
+				return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject)
+			}
+			return nil, err
+		}
+		return kv, nil
 	}}
 }
 
@@ -54,7 +67,7 @@ func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
 
 // Get is a key's current value, or nil when it has none.
 func (s State) Get(key string) (*stdio.StateEntry, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
 	defer cancel()
 	kv, err := s.store(ctx)
 	if err != nil {
@@ -73,7 +86,7 @@ func (s State) Get(key string) (*stdio.StateEntry, error) {
 // Create writes a key only when it has none, and returns the revision; a key that has one is refused as
 // stdio.ErrChanged.
 func (s State) Create(key string, value any) (uint64, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
 	defer cancel()
 	kv, err := s.store(ctx)
 	if err != nil {
@@ -96,7 +109,7 @@ func (s State) Create(key string, value any) (uint64, error) {
 // Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
 // is refused as stdio.ErrChanged.
 func (s State) Update(key string, value any, revision uint64) (uint64, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
 	defer cancel()
 	kv, err := s.store(ctx)
 	if err != nil {

go/process/state_integration_test.go

@@ -6,8 +6,8 @@ import (
 	"errors"
 	"strings"
 	"testing"
-	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
@@ -64,15 +64,18 @@ func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) {
 	}
 }
 
-// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes back
-// as an error naming the bucket. A denied publish is refused only as an async protocol error with no
-// reply (never a reply naming the refusal), so StateTimeout is shortened for this test alone.
+// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes
+// back as an error naming the bucket. The bucket exists (made by the admin) so the only reason Create
+// can fail is the missing grant, never "bucket not found"; and Conn's own ErrorHandler (conn.go
+// withRefusal/publishRefused) catches the denied publish at once, so this returns promptly rather than
+// after a timeout.
 func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
-	old := StateTimeout
-	StateTimeout = 2 * time.Second
-	t.Cleanup(func() { StateTimeout = old })
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
 
-	_, account := grantedSendProcess(t)
 	conn, err := Connect(account)
 	if err != nil {
 		t.Fatal(err)
@@ -80,9 +83,14 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
 	defer conn.Close()
 
 	s := conn.State("unrelated")
-	if _, err := s.Create("key", map[string]any{"v": 1}); err == nil {
+	_, err = s.Create("key", map[string]any{"v": 1})
+	if err == nil {
 		t.Fatal("writing to an ungranted bucket was not refused")
-	} else if !strings.Contains(err.Error(), "mesh-mail_unrelated") {
+	}
+	if !errors.Is(err, nats.ErrPermissionViolation) {
+		t.Errorf("the refusal %v does not wrap nats.ErrPermissionViolation", err)
+	}
+	if !strings.Contains(err.Error(), "mesh-mail_unrelated") {
 		t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error())
 	}
 }