[CHANGED] Gateway: Switch all accounts to interest-only mode

We are phasing out the optimistic-only mode. Servers accepting
inbound gateway connections will switch the accounts to interest-only
mode.

The servers with outbound gateway connection will check interest
and ignore the "optimistic" mode if it is known that the corresponding
inbound is going to switch the account to interest-only. This is
done using a boolean in the gateway INFO protocol.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

server/gateway.go

@@ -118,6 +118,14 @@ func (im GatewayInterestMode) String() string {
 	}
 }
 
+var gwDoNotForceInterestOnlyMode bool
+
+// GatewayDoNotForceInterestOnlyMode is used ONLY in tests.
+// DO NOT USE in normal code or if you embed the NATS Server.
+func GatewayDoNotForceInterestOnlyMode(doNotForce bool) {
+	gwDoNotForceInterestOnlyMode = doNotForce
+}
+
 type srvGateway struct {
 	totalQSubs int64 //total number of queue subs in all remote gateways (used with atomic operations)
 	sync.RWMutex
@@ -193,16 +201,21 @@ type gatewayCfg struct {
 // Struct for client's gateway related fields
 type gateway struct {
 	name       string
-	outbound   bool
 	cfg        *gatewayCfg
 	connectURL *url.URL          // Needed when sending CONNECT after receiving INFO from remote
 	outsim     *sync.Map         // Per-account subject interest (or no-interest) (outbound conn)
 	insim      map[string]*insie // Per-account subject no-interest sent or modeInterestOnly mode (inbound conn)
 
+	// This is an outbound GW connection
+	outbound bool
 	// Set/check in readLoop without lock. This is to know that an inbound has sent the CONNECT protocol first
 	connected bool
 	// Set to true if outbound is to a server that only knows about $GR, not $GNR
 	useOldPrefix bool
+	// If true, it indicates that the inbound side will switch any account to
+	// interest-only mode "immediately", so the outbound should disregard
+	// the optimistic mode when checking for interest.
+	interestOnlyMode bool
 }
 
 // Outbound subject interest entry.
@@ -510,6 +523,14 @@ func (s *Server) startGatewayAcceptLoop() {
 		GatewayNRP:   true,
 		Headers:      s.supportsHeaders(),
 	}
+	// Unless in some tests we want to keep the old behavior, we are now
+	// (since v2.9.0) indicate that this server will switch all accounts
+	// to InterestOnly mode when accepting an inbound or when a new
+	// account is fetched.
+	if !gwDoNotForceInterestOnlyMode {
+		info.GatewayIOM = true
+	}
+
 	// If we have selected a random port...
 	if port == 0 {
 		// Write resolved port back to options.
@@ -1038,6 +1059,7 @@ func (c *client) processGatewayInfo(info *Info) {
 			// from this INFO protocol and can sign it in the CONNECT we are
 			// going to send now.
 			c.mu.Lock()
+			c.gw.interestOnlyMode = info.GatewayIOM
 			c.sendGatewayConnect(opts)
 			c.Debugf("Gateway connect protocol sent to %q", gwName)
 			// Send INFO too
@@ -1122,8 +1144,9 @@ func (c *client) processGatewayInfo(info *Info) {
 		js := s.js
 		s.mu.Unlock()
 
-		// Switch JetStream accounts to interest-only mode.
-		if js != nil {
+		// If running in some tests, maintain the original behavior.
+		if gwDoNotForceInterestOnlyMode && js != nil {
+			// Switch JetStream accounts to interest-only mode.
 			var accounts []string
 			js.mu.Lock()
 			if len(js.accounts) > 0 {
@@ -1140,6 +1163,15 @@ func (c *client) processGatewayInfo(info *Info) {
 					}
 				}
 			}
+		} else if !gwDoNotForceInterestOnlyMode {
+			// Starting 2.9.0, we are phasing out the optimistic mode, so change
+			// all accounts to interest-only mode, unless instructed not to do so
+			// in some tests.
+			s.accounts.Range(func(_, v interface{}) bool {
+				acc := v.(*Account)
+				s.switchAccountToInterestMode(acc.GetName())
+				return true
+			})
 		}
 	}
 }
@@ -2050,17 +2082,19 @@ func (c *client) gatewayInterest(acc, subj string) (bool, *SublistResult) {
 	if accountInMap && ei == nil {
 		return false, nil
 	}
-	// Assume interest if account not in map.
-	psi := !accountInMap
+	// Assume interest if account not in map, unless we support
+	// only interest-only mode.
+	psi := !accountInMap && !c.gw.interestOnlyMode
 	var r *SublistResult
 	if accountInMap {
 		// If in map, check for subs interest with sublist.
 		e := ei.(*outsie)
 		e.RLock()
-		// We may be in transition to modeInterestOnly
+		// Unless each side has agreed on interest-only mode,
+		// we may be in transition to modeInterestOnly
 		// but until e.ni is nil, use it to know if we
 		// should suppress interest or not.
-		if e.ni != nil {
+		if !c.gw.interestOnlyMode && e.ni != nil {
 			if _, inMap := e.ni[subj]; !inMap {
 				psi = true
 			}

server/gateway_test.go

@@ -144,6 +144,60 @@ func checkForAccountNoInterest(t *testing.T, c *client, account string, expected
 	})
 }
 
+func checkGWInterestOnlyMode(t *testing.T, s *Server, outboundGWName, accName string) {
+	t.Helper()
+	checkGWInterestOnlyModeOrNotPresent(t, s, outboundGWName, accName, false)
+}
+
+func checkGWInterestOnlyModeOrNotPresent(t *testing.T, s *Server, outboundGWName, accName string, notPresentOk bool) {
+	t.Helper()
+	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+		gwc := s.getOutboundGatewayConnection(outboundGWName)
+		if gwc == nil {
+			return fmt.Errorf("No outbound gateway connection %q for server %v", outboundGWName, s)
+		}
+		gwc.mu.Lock()
+		defer gwc.mu.Unlock()
+		out, ok := gwc.gw.outsim.Load(accName)
+		if !ok {
+			if notPresentOk {
+				return nil
+			} else {
+				return fmt.Errorf("Server %v - outbound gateway connection %q: no account %q found in map",
+					s, outboundGWName, accName)
+			}
+		}
+		if out == nil {
+			return fmt.Errorf("Server %v - outbound gateway connection %q: interest map not found for account %q",
+				s, outboundGWName, accName)
+		} else if mode := out.(*outsie).mode; mode != InterestOnly {
+			return fmt.Errorf(
+				"Server %v - outbound gateway connection %q: account %q mode shoule be InterestOnly but is %v",
+				s, outboundGWName, accName, mode)
+		}
+		return nil
+	})
+}
+
+func checkGWInterestOnlyModeInterestOn(t *testing.T, s *Server, outboundGWName, accName, subject string) {
+	t.Helper()
+	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+		c := s.getOutboundGatewayConnection(outboundGWName)
+		outsiei, _ := c.gw.outsim.Load(accName)
+		if outsiei == nil {
+			return fmt.Errorf("Server %s - outbound gateway connection %q: no map entry found for account %q",
+				s, outboundGWName, accName)
+		}
+		outsie := outsiei.(*outsie)
+		r := outsie.sl.Match(subject)
+		if len(r.psubs) == 0 {
+			return fmt.Errorf("Server %s - outbound gateway connection %q - account %q: no subject interest for %q",
+				s, outboundGWName, accName, subject)
+		}
+		return nil
+	})
+}
+
 func waitCh(t *testing.T, ch chan bool, errTxt string) {
 	t.Helper()
 	select {
@@ -1666,6 +1720,9 @@ func setAccountUserPassInOptions(o *Options, accName, username, password string)
 }
 
 func TestGatewayAccountInterest(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	o2 := testDefaultOptionsForGateway("B")
 	// Add users to cause s2 to require auth. Will add an account with user later.
 	o2.Users = append([]*User(nil), &User{Username: "test", Password: "pwd"})
@@ -1841,6 +1898,9 @@ func TestGatewayAccountUnsub(t *testing.T) {
 }
 
 func TestGatewaySubjectInterest(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	o1 := testDefaultOptionsForGateway("A")
 	setAccountUserPassInOptions(o1, "$foo", "ivan", "password")
 	s1 := runGatewayServer(o1)
@@ -2497,6 +2557,9 @@ func TestGatewaySendQSubsOnGatewayConnect(t *testing.T) {
 }
 
 func TestGatewaySendRemoteQSubs(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	ob1 := testDefaultOptionsForGateway("B")
 	sb1 := runGatewayServer(ob1)
 	defer sb1.Shutdown()
@@ -3281,6 +3344,9 @@ func getInboundGatewayConnection(s *Server, name string) *client {
 }
 
 func TestGatewaySendAllSubs(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	gatewayMaxRUnsubBeforeSwitch = 100
 	defer func() { gatewayMaxRUnsubBeforeSwitch = defaultGatewayMaxRUnsubBeforeSwitch }()
 
@@ -3392,21 +3458,9 @@ func TestGatewaySendAllSubs(t *testing.T) {
 		if !switchedMode {
 			return fmt.Errorf("C has still not switched mode")
 		}
-		switchedMode = false
-		// Now check B outbound connection to C
-		c = sb.getOutboundGatewayConnection("C")
-		ei, _ := c.gw.outsim.Load(globalAccountName)
-		if ei != nil {
-			e := ei.(*outsie)
-			e.RLock()
-			switchedMode = e.ni == nil && e.mode == InterestOnly
-			e.RUnlock()
-		}
-		if !switchedMode {
-			return fmt.Errorf("C has still not switched mode")
-		}
 		return nil
 	})
+	checkGWInterestOnlyMode(t, sb, "C", globalAccountName)
 	wg.Wait()
 
 	// Check consCount and accsCount on C
@@ -3582,6 +3636,9 @@ func TestGatewayRaceOnClose(t *testing.T) {
 // Similar to TestNewRoutesServiceImport but with 2 GW servers instead
 // of a cluster of 2 servers.
 func TestGatewayServiceImport(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	oa := testDefaultOptionsForGateway("A")
 	setAccountUserPassInOptions(oa, "$foo", "clientA", "password")
 	setAccountUserPassInOptions(oa, "$bar", "yyyyyyy", "password")
@@ -3809,21 +3866,7 @@ func TestGatewayServiceImport(t *testing.T) {
 	}
 	natsFlush(t, clientB)
 
-	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		c := sb.getOutboundGatewayConnection("A")
-		outsiei, _ := c.gw.outsim.Load("$foo")
-		if outsiei == nil {
-			return fmt.Errorf("Nothing found for $foo")
-		}
-		outsie := outsiei.(*outsie)
-		outsie.RLock()
-		mode := outsie.mode
-		outsie.RUnlock()
-		if mode != InterestOnly {
-			return fmt.Errorf("Should have switched to interest only mode")
-		}
-		return nil
-	})
+	checkGWInterestOnlyMode(t, sb, "A", "$foo")
 
 	// Go back to clientB on $bar.
 	clientB.Close()
@@ -3839,19 +3882,7 @@ func TestGatewayServiceImport(t *testing.T) {
 
 	// Sine it is interest-only, B should receive an interest
 	// on $foo test.request
-	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		c := sb.getOutboundGatewayConnection("A")
-		outsiei, _ := c.gw.outsim.Load("$foo")
-		if outsiei == nil {
-			return fmt.Errorf("Nothing found for $foo")
-		}
-		outsie := outsiei.(*outsie)
-		r := outsie.sl.Match("test.request")
-		if len(r.psubs) != 1 {
-			return fmt.Errorf("No registered interest on test.request")
-		}
-		return nil
-	})
+	checkGWInterestOnlyModeInterestOn(t, sb, "A", "$foo", "test.request")
 
 	// Send the request from clientB on foo.request,
 	natsPubReq(t, clientB, "foo.request", "reply", []byte("hi"))
@@ -3893,6 +3924,9 @@ func TestGatewayServiceImport(t *testing.T) {
 }
 
 func TestGatewayServiceImportWithQueue(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	oa := testDefaultOptionsForGateway("A")
 	setAccountUserPassInOptions(oa, "$foo", "clientA", "password")
 	setAccountUserPassInOptions(oa, "$bar", "yyyyyyy", "password")
@@ -4124,21 +4158,7 @@ func TestGatewayServiceImportWithQueue(t *testing.T) {
 	}
 	natsFlush(t, clientB)
 
-	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		c := sb.getOutboundGatewayConnection("A")
-		outsiei, _ := c.gw.outsim.Load("$foo")
-		if outsiei == nil {
-			return fmt.Errorf("Nothing found for $foo")
-		}
-		outsie := outsiei.(*outsie)
-		outsie.RLock()
-		mode := outsie.mode
-		outsie.RUnlock()
-		if mode != InterestOnly {
-			return fmt.Errorf("Should have switched to interest only mode")
-		}
-		return nil
-	})
+	checkGWInterestOnlyMode(t, sb, "A", "$foo")
 
 	// Go back to clientB on $bar.
 	clientB.Close()
@@ -4154,19 +4174,7 @@ func TestGatewayServiceImportWithQueue(t *testing.T) {
 
 	// Sine it is interest-only, B should receive an interest
 	// on $foo test.request
-	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		c := sb.getOutboundGatewayConnection("A")
-		outsiei, _ := c.gw.outsim.Load("$foo")
-		if outsiei == nil {
-			return fmt.Errorf("Nothing found for $foo")
-		}
-		outsie := outsiei.(*outsie)
-		r := outsie.sl.Match("test.request")
-		if len(r.psubs) != 1 {
-			return fmt.Errorf("No registered interest on test.request")
-		}
-		return nil
-	})
+	checkGWInterestOnlyModeInterestOn(t, sb, "A", "$foo", "test.request")
 
 	// Send the request from clientB on foo.request,
 	natsPubReq(t, clientB, "foo.request", "reply", []byte("hi"))
@@ -4517,24 +4525,7 @@ func TestGatewayServiceImportComplexSetup(t *testing.T) {
 	}
 	natsFlush(t, clientA)
 	// Wait for B2 to switch to interest-only
-	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		c := sa2.getOutboundGatewayConnection("B")
-		if c == nil {
-			return fmt.Errorf("No outbound to B2")
-		}
-		outsiei, _ := c.gw.outsim.Load("$foo")
-		if outsiei == nil {
-			return fmt.Errorf("Nothing for $foo")
-		}
-		outsie := outsiei.(*outsie)
-		outsie.RLock()
-		mode := outsie.mode
-		outsie.RUnlock()
-		if mode != InterestOnly {
-			return fmt.Errorf("Not in interest-only mode yet")
-		}
-		return nil
-	})
+	checkGWInterestOnlyMode(t, sa2, "B", "$foo")
 
 	subA = natsSubSync(t, clientA, "test.request")
 	natsFlush(t, clientA)
@@ -4890,24 +4881,7 @@ func TestGatewayServiceExportWithWildcards(t *testing.T) {
 			natsFlush(t, clientA)
 
 			// Wait for B2 to switch to interest-only
-			checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-				c := sa2.getOutboundGatewayConnection("B")
-				if c == nil {
-					return fmt.Errorf("No outbound to B2")
-				}
-				outsiei, _ := c.gw.outsim.Load("$foo")
-				if outsiei == nil {
-					return fmt.Errorf("Nothing for $foo")
-				}
-				outsie := outsiei.(*outsie)
-				outsie.RLock()
-				mode := outsie.mode
-				outsie.RUnlock()
-				if mode != InterestOnly {
-					return fmt.Errorf("Not in interest-only mode yet")
-				}
-				return nil
-			})
+			checkGWInterestOnlyMode(t, sa2, "B", "$foo")
 
 			subA = natsSubSync(t, clientA, "ngs.update.*")
 			natsFlush(t, clientA)
@@ -5541,6 +5515,9 @@ func TestGatewayClientsDontReceiveMsgsOnGWPrefix(t *testing.T) {
 }
 
 func TestGatewayNoAccInterestThenQSubThenRegularSub(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -5600,6 +5577,9 @@ func TestGatewayNoAccInterestThenQSubThenRegularSub(t *testing.T) {
 // Similar to TestGatewayNoAccInterestThenQSubThenRegularSub but simulate
 // older incorrect behavior.
 func TestGatewayHandleUnexpectedASubUnsub(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -5735,6 +5715,9 @@ func (l *captureGWInterestSwitchLogger) Debugf(format string, args ...interface{
 }
 
 func TestGatewayLogAccountInterestModeSwitch(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -5782,21 +5765,7 @@ func TestGatewayLogAccountInterestModeSwitch(t *testing.T) {
 		return nil
 	})
 
-	gwB := sa.getOutboundGatewayConnection("B")
-	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
-		ei, ok := gwB.gw.outsim.Load(globalAccountName)
-		if !ok || ei == nil {
-			return fmt.Errorf("not switched yet")
-		}
-		e := ei.(*outsie)
-		e.RLock()
-		mode := e.mode
-		e.RUnlock()
-		if mode != InterestOnly {
-			return fmt.Errorf("not in interest mode only yet")
-		}
-		return nil
-	})
+	checkGWInterestOnlyMode(t, sa, "B", globalAccountName)
 
 	checkLog := func(t *testing.T, l *captureGWInterestSwitchLogger) {
 		t.Helper()
@@ -5835,6 +5804,9 @@ func TestGatewayLogAccountInterestModeSwitch(t *testing.T) {
 }
 
 func TestGatewayAccountInterestModeSwitchOnlyOncePerAccount(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -6780,3 +6752,72 @@ func TestGatewayNoPanicOnStartupWithMonitoring(t *testing.T) {
 	s.Shutdown()
 	wg.Wait()
 }
+
+func TestGatewaySwitchToInterestOnlyModeImmediately(t *testing.T) {
+	o2 := testDefaultOptionsForGateway("B")
+	// Add users to cause s2 to require auth. Will add an account with user later.
+	o2.Users = append([]*User(nil), &User{Username: "test", Password: "pwd"})
+	s2 := runGatewayServer(o2)
+	defer s2.Shutdown()
+
+	o1 := testGatewayOptionsFromToWithServers(t, "A", "B", s2)
+	setAccountUserPassInOptions(o1, "$foo", "ivan", "password")
+	s1 := runGatewayServer(o1)
+	defer s1.Shutdown()
+
+	waitForOutboundGateways(t, s1, 1, time.Second)
+	waitForOutboundGateways(t, s2, 1, time.Second)
+
+	s1Url := fmt.Sprintf("nats://ivan:password@127.0.0.1:%d", o1.Port)
+	nc := natsConnect(t, s1Url)
+	defer nc.Close()
+	natsPub(t, nc, "foo", []byte("hello"))
+	natsFlush(t, nc)
+
+	checkCount := func(t *testing.T, c *client, expected int) {
+		t.Helper()
+		c.mu.Lock()
+		out := c.outMsgs
+		c.mu.Unlock()
+		if int(out) != expected {
+			t.Fatalf("Expected %d message(s) to be sent over, got %v", expected, out)
+		}
+	}
+	// No message should be sent
+	gwcb := s1.getOutboundGatewayConnection("B")
+	checkCount(t, gwcb, 0)
+
+	// Check that we are in interest-only mode, but in this case, since s2 does
+	// have the account, we should have the account not even present in the map.
+	checkGWInterestOnlyModeOrNotPresent(t, s1, "B", "$foo", true)
+
+	// Add account to S2 and a client.
+	s2FooAcc, err := s2.RegisterAccount("$foo")
+	if err != nil {
+		t.Fatalf("Error registering account: %v", err)
+	}
+	s2.mu.Lock()
+	s2.users["ivan"] = &User{Account: s2FooAcc, Username: "ivan", Password: "password"}
+	s2.mu.Unlock()
+	s2Url := fmt.Sprintf("nats://ivan:password@127.0.0.1:%d", o2.Port)
+	ncS2 := natsConnect(t, s2Url)
+	defer ncS2.Close()
+	natsSubSync(t, ncS2, "asub")
+	// This time we will have the account in the map and it will be interest-only
+	checkGWInterestOnlyMode(t, s1, "B", "$foo")
+
+	// Now publish a message, still should not go because the sub is on "asub"
+	natsPub(t, nc, "foo", []byte("hello"))
+	natsFlush(t, nc)
+	checkCount(t, gwcb, 0)
+
+	natsSubSync(t, ncS2, "foo")
+	natsFlush(t, ncS2)
+
+	checkGWInterestOnlyModeInterestOn(t, s1, "B", "$foo", "foo")
+
+	// Publish on foo
+	natsPub(t, nc, "foo", []byte("hello"))
+	natsFlush(t, nc)
+	checkCount(t, gwcb, 1)
+}

server/jetstream_super_cluster_test.go

@@ -22,6 +22,8 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"net/http"
+	"net/http/httptest"
 	"reflect"
 	"strings"
 	"sync"
@@ -29,7 +31,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 )
 
 func TestJetStreamSuperClusterMetaPlacement(t *testing.T) {
@@ -413,6 +417,9 @@ func TestJetStreamSuperClusterPeerReassign(t *testing.T) {
 }
 
 func TestJetStreamSuperClusterInterestOnlyMode(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	template := `
 		listen: 127.0.0.1:-1
 		server_name: %s
@@ -3611,3 +3618,210 @@ func TestJetStreamSuperClusterStreamCathupLongRTT(t *testing.T) {
 		return nil
 	})
 }
+
+func TestJetStreamSuperClusterMixedModeSwitchToInterestOnlyStaticConfig(t *testing.T) {
+	tmpl := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: { domain: ngs, max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		leaf: { listen: 127.0.0.1:-1 }
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+		accounts {
+			ONE {
+				users = [  { user: "one", pass: "pwd" } ]
+				jetstream: enabled
+			}
+			TWO { users = [  { user: "two", pass: "pwd" } ] }
+			$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+		}
+	`
+	sc := createJetStreamSuperClusterWithTemplateAndModHook(t, tmpl, 5, 3,
+		func(serverName, clusterName, storeDir, conf string) string {
+			sname := serverName[strings.Index(serverName, "-")+1:]
+			switch sname {
+			case "S4", "S5":
+				conf = strings.ReplaceAll(conf, "jetstream: { ", "#jetstream: { ")
+			default:
+				conf = strings.ReplaceAll(conf, "leaf: { ", "#leaf: { ")
+			}
+			return conf
+		}, nil)
+	defer sc.shutdown()
+
+	// Connect our client to a non JS server
+	c := sc.randomCluster()
+	var s *Server
+	for _, as := range c.servers {
+		if !as.JetStreamEnabled() {
+			s = as
+			break
+		}
+	}
+	if s == nil {
+		t.Fatal("Did not find a non JS server!")
+	}
+	nc, js := jsClientConnect(t, s, nats.UserInfo("one", "pwd"))
+	defer nc.Close()
+
+	// Just create a stream and then make sure that all gateways have switched
+	// to interest-only mode.
+	si, err := js.AddStream(&nats.StreamConfig{Name: "interest", Replicas: 3})
+	require_NoError(t, err)
+
+	sc.waitOnStreamLeader("ONE", "interest")
+
+	check := func(accName string) {
+		t.Helper()
+		for _, c := range sc.clusters {
+			for _, s := range c.servers {
+				// Check only JS servers outbound GW connections
+				if !s.JetStreamEnabled() {
+					continue
+				}
+				opts := s.getOpts()
+				for _, gw := range opts.Gateway.Gateways {
+					if gw.Name == opts.Gateway.Name {
+						continue
+					}
+					checkGWInterestOnlyMode(t, s, gw.Name, accName)
+				}
+			}
+		}
+	}
+	// Starting v2.9.0, all accounts should be switched to interest-only mode
+	check("ONE")
+	check("TWO")
+
+	var gwsa [16]*client
+	gws := gwsa[:0]
+
+	s = sc.serverByName(si.Cluster.Leader)
+	// Get the GW outbound connections
+	s.getOutboundGatewayConnections(&gws)
+	for _, gwc := range gws {
+		gwc.mu.Lock()
+		gwc.nc.Close()
+		gwc.mu.Unlock()
+	}
+	waitForOutboundGateways(t, s, 2, 5*time.Second)
+	check("ONE")
+	check("TWO")
+}
+
+func TestJetStreamSuperClusterMixedModeSwitchToInterestOnlyOperatorConfig(t *testing.T) {
+	kp, _ := nkeys.FromSeed(oSeed)
+
+	skp, _ := nkeys.CreateAccount()
+	spub, _ := skp.PublicKey()
+	nac := jwt.NewAccountClaims(spub)
+	sjwt, err := nac.Encode(kp)
+	require_NoError(t, err)
+
+	akp, _ := nkeys.CreateAccount()
+	apub, _ := akp.PublicKey()
+	nac = jwt.NewAccountClaims(apub)
+	// Set some limits to enable JS.
+	nac.Limits.JetStreamLimits.DiskStorage = 1024 * 1024
+	nac.Limits.JetStreamLimits.Streams = 10
+	ajwt, err := nac.Encode(kp)
+	require_NoError(t, err)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, spub) {
+			w.Write([]byte(sjwt))
+		} else {
+			w.Write([]byte(ajwt))
+		}
+	}))
+	defer ts.Close()
+
+	operator := fmt.Sprintf(`
+		operator: %s
+		resolver: URL("%s/ngs/v1/accounts/jwt/")
+	`, ojwt, ts.URL)
+
+	tmpl := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: { domain: ngs, max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		leaf: { listen: 127.0.0.1:-1 }
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+	` + operator
+	sc := createJetStreamSuperClusterWithTemplateAndModHook(t, tmpl, 5, 3,
+		func(serverName, clusterName, storeDir, conf string) string {
+			conf = strings.ReplaceAll(conf, "system_account: \"$SYS\"", fmt.Sprintf("system_account: \"%s\"", spub))
+			sname := serverName[strings.Index(serverName, "-")+1:]
+			switch sname {
+			case "S4", "S5":
+				conf = strings.ReplaceAll(conf, "jetstream: { ", "#jetstream: { ")
+			default:
+				conf = strings.ReplaceAll(conf, "leaf: { ", "#leaf: { ")
+			}
+			return conf
+		}, nil)
+	defer sc.shutdown()
+
+	// Connect our client to a non JS server
+	c := sc.randomCluster()
+	var s *Server
+	for _, as := range c.servers {
+		if !as.JetStreamEnabled() {
+			s = as
+			break
+		}
+	}
+	if s == nil {
+		t.Fatal("Did not find a non JS server!")
+	}
+	nc, js := jsClientConnect(t, s, createUserCreds(t, nil, akp))
+	defer nc.Close()
+
+	// Just create a stream and then make sure that all gateways have switched
+	// to interest-only mode.
+	si, err := js.AddStream(&nats.StreamConfig{Name: "interest", Replicas: 3})
+	require_NoError(t, err)
+
+	sc.waitOnStreamLeader(apub, "interest")
+
+	check := func(s *Server) {
+		opts := s.getOpts()
+		for _, gw := range opts.Gateway.Gateways {
+			if gw.Name == opts.Gateway.Name {
+				continue
+			}
+			checkGWInterestOnlyMode(t, s, gw.Name, apub)
+		}
+	}
+	s = sc.serverByName(si.Cluster.Leader)
+	check(s)
+
+	// Let's cause a leadership change and verify that it still works.
+	_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "interest"), nil, time.Second)
+	require_NoError(t, err)
+	sc.waitOnStreamLeader(apub, "interest")
+
+	si, err = js.StreamInfo("interest")
+	require_NoError(t, err)
+	s = sc.serverByName(si.Cluster.Leader)
+	check(s)
+
+	var gwsa [16]*client
+	gws := gwsa[:0]
+	// Get the GW outbound connections
+	s.getOutboundGatewayConnections(&gws)
+	for _, gwc := range gws {
+		gwc.mu.Lock()
+		gwc.nc.Close()
+		gwc.mu.Unlock()
+	}
+	waitForOutboundGateways(t, s, 2, 5*time.Second)
+	check(s)
+}

server/monitor_test.go

@@ -3168,6 +3168,9 @@ func TestMonitorGatewayz(t *testing.T) {
 }
 
 func TestMonitorGatewayzAccounts(t *testing.T) {
+	GatewayDoNotForceInterestOnlyMode(true)
+	defer GatewayDoNotForceInterestOnlyMode(false)
+
 	resetPreviousHTTPConnections()
 
 	// Create bunch of Accounts

server/server.go

@@ -98,6 +98,7 @@ type Info struct {
 	GatewayCmd        byte     `json:"gateway_cmd,omitempty"`         // Command code for the receiving server to know what to do
 	GatewayCmdPayload []byte   `json:"gateway_cmd_payload,omitempty"` // Command payload when needed
 	GatewayNRP        bool     `json:"gateway_nrp,omitempty"`         // Uses new $GNR. prefix for mapped replies
+	GatewayIOM        bool     `json:"gateway_iom,omitempty"`         // Indicate that all accounts will be switched to InterestOnly mode "right away"
 
 	// LeafNode Specific
 	LeafNodeURLs  []string `json:"leafnode_urls,omitempty"`  // LeafNode URLs that the server can reconnect to.
@@ -1413,6 +1414,12 @@ func (s *Server) registerAccountNoLock(acc *Account) *Account {
 	// Can not have server lock here.
 	s.mu.Unlock()
 	s.registerSystemImports(acc)
+	// Starting 2.9.0, we are phasing out the optimistic mode, so change
+	// the account to interest-only mode (except if instructed not to do
+	// it in some tests).
+	if s.gateway.enabled && !gwDoNotForceInterestOnlyMode {
+		s.switchAccountToInterestMode(acc.GetName())
+	}
 	s.mu.Lock()
 
 	return nil

test/gateway_test.go

@@ -82,6 +82,9 @@ func expectNumberOfProtos(t *testing.T, expFn expectFun, proto *regexp.Regexp, e
 }
 
 func TestGatewayAccountInterest(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -159,6 +162,9 @@ func TestGatewayAccountInterest(t *testing.T) {
 }
 
 func TestGatewaySubjectInterest(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	fooAcc := server.NewAccount("$foo")
 	ob.Accounts = []*server.Account{fooAcc}
@@ -301,6 +307,9 @@ func TestGatewaySubjectInterest(t *testing.T) {
 }
 
 func TestGatewayQueue(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	fooAcc := server.NewAccount("$foo")
 	ob.Accounts = []*server.Account{fooAcc}
@@ -403,6 +412,9 @@ func TestGatewayQueue(t *testing.T) {
 }
 
 func TestGatewaySendAllSubs(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -496,6 +508,9 @@ func TestGatewayNoPanicOnBadProtocol(t *testing.T) {
 }
 
 func TestGatewayNoAccUnsubAfterQSub(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -531,6 +546,9 @@ func TestGatewayNoAccUnsubAfterQSub(t *testing.T) {
 }
 
 func TestGatewayErrorOnRSentFromOutbound(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	sb := runGatewayServer(ob)
 	defer sb.Shutdown()
@@ -675,6 +693,9 @@ func TestGatewayTLSMixedIPAndDNS(t *testing.T) {
 }
 
 func TestGatewayAdvertiseInCluster(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob1 := testDefaultOptionsForGateway("B")
 	ob1.Cluster.Name = "B"
 	ob1.Cluster.Host = "127.0.0.1"
@@ -770,6 +791,9 @@ func TestGatewayAuthTimeout(t *testing.T) {
 }
 
 func TestGatewayFirstPingGoesAfterConnect(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	ob := testDefaultOptionsForGateway("B")
 	// For this test, we want the first ping to NOT be disabled.
 	ob.DisableShortFirstPing = false

test/leafnode_test.go

@@ -2665,6 +2665,9 @@ func TestLeafNodeSwitchGatewayToInterestModeOnly(t *testing.T) {
 
 // route connections to simulate.
 func TestLeafNodeResetsMSGProto(t *testing.T) {
+	server.GatewayDoNotForceInterestOnlyMode(true)
+	defer server.GatewayDoNotForceInterestOnlyMode(false)
+
 	opts := testDefaultOptionsForLeafNodes()
 	opts.Cluster.Name = "xyz"
 	opts.Cluster.Host = opts.Host