[FIXED] Gateway RS+/- blocks on account fetch (#7449)

The methods `processGatewayRSub` and `processGatewayRUnsub` can call
into `updateInterestForAccountOnGateway`, which in turn calls
`s.LookupAccount`. If the account is not known it will be fetched if
there's an account resolver. But, since this is done while in the read
loop of the gateway, this blocks not only receiving the response for the
account fetch (when using a NATS resolver) but also any other operation
like PING/PONG, which then could lead to a stale connection.

The fix proposed by this PR is to not allow fetching the account inline
of the gateway read loop.

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Author: neilalexander

server/gateway_test.go

@@ -33,6 +33,7 @@ import (
 
 	"github.com/nats-io/nats-server/v2/logger"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 	"golang.org/x/crypto/ocsp"
 
 	. "github.com/nats-io/nats-server/v2/internal/ocsp"
@@ -7557,3 +7558,59 @@ func TestGatewayConfigureWriteDeadline(t *testing.T) {
 		require_Equal(t, r.out.wdl, 5*time.Second)
 	})
 }
+
+func TestGatewayProcessRSubNoBlockingAccountFetch(t *testing.T) {
+	createAccountPubKey := func() string {
+		kp, err := nkeys.CreateAccount()
+		require_NoError(t, err)
+		pubkey, err := kp.PublicKey()
+		require_NoError(t, err)
+		return pubkey
+	}
+	sysPub := createAccountPubKey()
+	accPub := createAccountPubKey()
+	dir := t.TempDir()
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		server_name: srv
+		operator: %s
+		system_account: %s
+		resolver: {
+			type: cache
+			dir: '%s'
+			timeout: "2s"
+		}
+		gateway: {
+			name: "clust-B"
+			listen: 127.0.0.1:-1
+		}
+       `, ojwt, sysPub, dir)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	// Set up a mock gateway client.
+	c := s.createInternalAccountClient()
+	c.mu.Lock()
+	c.gw = &gateway{}
+	c.gw.outsim = &sync.Map{}
+	c.nc = &net.IPConn{}
+	c.mu.Unlock()
+
+	// Receiving a R+ should not be blocking, since we're in the gateway's readLoop.
+	start := time.Now()
+	require_NoError(t, c.processGatewayRSub(fmt.Appendf(nil, "%s subj queue 0", accPub)))
+	c.mu.Lock()
+	subs := len(c.subs)
+	c.mu.Unlock()
+	require_Len(t, subs, 1)
+	require_LessThan(t, time.Since(start), 100*time.Millisecond)
+
+	// Receiving a R- should not be blocking, since we're in the gateway's readLoop.
+	start = time.Now()
+	require_NoError(t, c.processGatewayRUnsub(fmt.Appendf(nil, "%s subj queue", accPub)))
+	c.mu.Lock()
+	subs = len(c.subs)
+	c.mu.Unlock()
+	require_Len(t, subs, 0)
+	require_LessThan(t, time.Since(start), 100*time.Millisecond)
+}

server/jetstream_super_cluster_test.go

@@ -4411,7 +4411,7 @@ func TestJetStreamSuperClusterMixedModeSwitchToInterestOnlyOperatorConfig(t *tes
 			if gw.Name == opts.Gateway.Name {
 				continue
 			}
-			checkGWInterestOnlyMode(t, s, gw.Name, apub)
+			checkGWInterestOnlyModeOrNotPresent(t, s, gw.Name, apub, true)
 		}
 	}
 	s = sc.serverByName(si.Cluster.Leader)

server/leafnode.go

@@ -2426,7 +2426,8 @@ func (s *Server) initLeafNodeSmapAndSendSubs(c *client) {
 
 // updateInterestForAccountOnGateway called from gateway code when processing RS+ and RS-.
 func (s *Server) updateInterestForAccountOnGateway(accName string, sub *subscription, delta int32) {
-	acc, err := s.LookupAccount(accName)
+	// Since we're in the gateway's readLoop, and we would otherwise block, don't allow fetching.
+	acc, err := s.lookupOrFetchAccount(accName, false)
 	if acc == nil || err != nil {
 		s.Debugf("No or bad account for %q, failed to update interest from gateway", accName)
 		return

server/server.go

@@ -2051,6 +2051,13 @@ func (s *Server) setRouteInfo(acc *Account) {
 // associated with an account name.
 // Lock MUST NOT be held upon entry.
 func (s *Server) lookupAccount(name string) (*Account, error) {
+	return s.lookupOrFetchAccount(name, true)
+}
+
+// lookupOrFetchAccount is a function to return the account structure
+// associated with an account name.
+// Lock MUST NOT be held upon entry.
+func (s *Server) lookupOrFetchAccount(name string, fetch bool) (*Account, error) {
 	var acc *Account
 	if v, ok := s.accounts.Load(name); ok {
 		acc = v.(*Account)
@@ -2060,7 +2067,7 @@ func (s *Server) lookupAccount(name string) (*Account, error) {
 		// return the latest information from the resolver.
 		if acc.IsExpired() {
 			s.Debugf("Requested account [%s] has expired", name)
-			if s.AccountResolver() != nil {
+			if s.AccountResolver() != nil && fetch {
 				if err := s.updateAccount(acc); err != nil {
 					// This error could mask expired, so just return expired here.
 					return nil, ErrAccountExpired
@@ -2072,7 +2079,7 @@ func (s *Server) lookupAccount(name string) (*Account, error) {
 		return acc, nil
 	}
 	// If we have a resolver see if it can fetch the account.
-	if s.AccountResolver() == nil {
+	if s.AccountResolver() == nil || !fetch {
 		return nil, ErrMissingAccount
 	}
 	return s.fetchAccount(name)