Repository: nats-io/nats-server
Pull request #4193 (2023-05-24 19:12:03)

When in async mode make sure to not flush or create new files after close.
Some async tests were showing artifacts being left after a message block was being closed.

Signed-off-by: Derek Collison <derek@nats.io>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 1a97c942..08eb1a87 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -3119,8 +3119,8 @@ func (mb *msgBlock) spinUpFlushLoop() {
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
 
-	// Are we already running?
-	if mb.flusher {
+	// Are we already running or closed?
+	if mb.flusher || mb.closed {
 		return
 	}
 	mb.flusher = true
@@ -3867,12 +3867,14 @@ func (mb *msgBlock) pendingWriteSize() int {
 	if mb == nil {
 		return 0
 	}
-	var pending int
+
 	mb.mu.RLock()
-	if mb.mfd != nil && mb.cache != nil {
+	defer mb.mu.RUnlock()
+
+	var pending int
+	if !mb.closed && mb.mfd != nil && mb.cache != nil {
 		pending = len(mb.cache.buf) - int(mb.cache.wp)
 	}
-	mb.mu.RUnlock()
 	return pending
 }
 
@@ -4615,6 +4617,7 @@ var (
 	errNoEncryption  = errors.New("encryption not enabled")
 	errBadKeySize    = errors.New("encryption bad key size")
 	errNoMsgBlk      = errors.New("no message block")
+	errMsgBlkClosed  = errors.New("message block is closed")
 	errMsgBlkTooBig  = errors.New("message block size exceeded int capacity")
 	errUnknownCipher = errors.New("unknown cipher")
 	errDIOStalled    = errors.New("IO is stalled")
@@ -5115,6 +5118,10 @@ func (mb *msgBlock) writeIndexInfo() error {
 // Write index info to the appropriate file.
 // Filestore lock and mb lock should be held.
 func (mb *msgBlock) writeIndexInfoLocked() error {
+	if mb.closed {
+		return errMsgBlkClosed
+	}
+
 	// HEADER: magic version msgs bytes fseq fts lseq lts ndel checksum
 	// Make large enough to hold almost all possible maximum interior delete scenarios.
 	var hdr [42 * 1024]byte
@@ -6401,7 +6408,6 @@ func (mb *msgBlock) close(sync bool) {
 	if mb.closed {
 		return
 	}
-	mb.closed = true
 
 	// Stop cache expiration timer.
 	if mb.ctmr != nil {
@@ -6435,6 +6441,8 @@ func (mb *msgBlock) close(sync bool) {
 	}
 	mb.mfd = nil
 	mb.ifd = nil
+	// Mark as closed.
+	mb.closed = true
 }
 
 func (fs *fileStore) closeAllMsgBlocks(sync bool) {
diff --git a/server/reload.go b/server/reload.go
index 0f18fce7..c7c7cbe7 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -2029,8 +2029,11 @@ func (s *Server) clientHasMovedToDifferentAccount(c *client) bool {
 // import subjects.
 func (s *Server) reloadClusterPermissions(oldPerms *RoutePermissions) {
 	s.mu.Lock()
-	newPerms := s.getOpts().Cluster.Permissions
-	routes := make(map[uint64]*client, s.numRoutes())
+	var (
+		infoJSON []byte
+		newPerms = s.getOpts().Cluster.Permissions
+		routes   = make(map[uint64]*client, s.numRoutes())
+	)
 	// Get all connected routes
 	s.forEachRoute(func(route *client) {
 		route.mu.Lock()
@@ -2045,7 +2048,8 @@ func (s *Server) reloadClusterPermissions(oldPerms *RoutePermissions) {
 		s.routeInfo.Import = newPerms.Import
 		s.routeInfo.Export = newPerms.Export
 	}
-	infoJSON := generateInfoJSON(&s.routeInfo)
+	// Copy the current route's INFO struct. We will need to modify it per-account
+	routeInfo := s.routeInfo
 	s.mu.Unlock()
 
 	// Close connections for routes that don't understand async INFO.
@@ -2071,46 +2075,17 @@ func (s *Server) reloadClusterPermissions(oldPerms *RoutePermissions) {
 	newPermsTester := &client{}
 	newPermsTester.setRoutePermissions(newPerms)
 
-	var (
-		_localSubs       [4096]*subscription
-		subsNeedSUB      = map[*client][]*subscription{}
-		subsNeedUNSUB    = map[*client][]*subscription{}
-		deleteRoutedSubs []*subscription
-	)
-
-	getRouteForAccount := func(accName string, poolIdx int) *client {
-		for _, r := range routes {
-			r.mu.Lock()
-			ok := (poolIdx >= 0 && poolIdx == r.route.poolIdx) || (string(r.route.accName) == accName) || r.route.noPool
-			r.mu.Unlock()
-			if ok {
-				return r
-			}
-		}
-		return nil
-	}
-
-	// First set the new permissions on all routes.
-	for _, route := range routes {
-		route.mu.Lock()
-		route.setRoutePermissions(newPerms)
-		route.mu.Unlock()
-	}
-
-	// Then, go over all accounts and gather local subscriptions that need to be
-	// sent over as SUB or removed as UNSUB, and routed subscriptions that need
-	// to be dropped due to export permissions.
-	s.accounts.Range(func(_, v interface{}) bool {
-		acc := v.(*Account)
-		acc.mu.RLock()
-		accName, sl, poolIdx := acc.Name, acc.sl, acc.routePoolIdx
-		acc.mu.RUnlock()
-		// Get the route handling this account. If no route or sublist, bail out.
-		route := getRouteForAccount(accName, poolIdx)
-		if route == nil || sl == nil {
-			return true
-		}
-		localSubs := _localSubs[:0]
+	// For a given account and list of remotes, will send an INFO protocol so
+	// that remote updates its route permissions and will also send the RS+ or
+	// RS- for subscriptions that become or are no longer permitted.
+	update := func(accName string, sl *Sublist, remotes []*client) {
+		var (
+			_localSubs       [4096]*subscription
+			localSubs        = _localSubs[:0]
+			subsNeedSUB      []*subscription
+			subsNeedUNSUB    []*subscription
+			deleteRoutedSubs []*subscription
+		)
 		sl.localSubs(&localSubs, false)
 
 		// Go through all local subscriptions
@@ -2122,49 +2097,64 @@ func (s *Server) reloadClusterPermissions(oldPerms *RoutePermissions) {
 			if canImportNow {
 				// If we could not before, then will need to send a SUB protocol.
 				if !couldImportThen {
-					subsNeedSUB[route] = append(subsNeedSUB[route], sub)
+					subsNeedSUB = append(subsNeedSUB, sub)
 				}
 			} else if couldImportThen {
 				// We were previously able to import this sub, but now
 				// we can't so we need to send an UNSUB protocol
-				subsNeedUNSUB[route] = append(subsNeedUNSUB[route], sub)
+				subsNeedUNSUB = append(subsNeedUNSUB, sub)
 			}
 		}
-		deleteRoutedSubs = deleteRoutedSubs[:0]
-		route.mu.Lock()
-		for key, sub := range route.subs {
-			if an := strings.Fields(key)[0]; an != accName {
-				continue
-			}
-			// If we can't export, we need to drop the subscriptions that
-			// we have on behalf of this route.
-			subj := string(sub.subject)
-			if !route.canExport(subj) {
-				delete(route.subs, string(sub.sid))
-				deleteRoutedSubs = append(deleteRoutedSubs, sub)
+
+		for _, route := range remotes {
+			// We do this manually here, not invoke generateRouteInfoJSON()
+			routeInfo.RemoteAccount = accName
+			infoJSON = generateInfoJSON(&routeInfo)
+
+			route.mu.Lock()
+			route.setRoutePermissions(newPerms)
+			for _, sub := range route.subs {
+				// If we can't export, we need to drop the subscriptions that
+				// we have on behalf of this route.
+				subj := string(sub.subject)
+				if !route.canExport(subj) {
+					delete(route.subs, string(sub.sid))
+					deleteRoutedSubs = append(deleteRoutedSubs, sub)
+				}
 			}
+			// Send an update INFO, which will allow remote server to show
+			// our current route config in monitoring and resend subscriptions
+			// that we now possibly allow with a change of Export permissions.
+			route.enqueueProto(infoJSON)
+			// Now send SUB and UNSUB protocols as needed.
+			route.sendRouteSubProtos(subsNeedSUB, false, nil)
+			route.sendRouteUnSubProtos(subsNeedUNSUB, false, nil)
+			route.mu.Unlock()
 		}
-		route.mu.Unlock()
 		// Remove as a batch all the subs that we have removed from each route.
 		sl.RemoveBatch(deleteRoutedSubs)
-		return true
-	})
+	}
 
-	// Send an update INFO, which will allow remote server to show
-	// our current route config in monitoring and resend subscriptions
-	// that we now possibly allow with a change of Export permissions.
-	for _, route := range routes {
-		route.mu.Lock()
-		route.enqueueProto(infoJSON)
-		// Now send SUB and UNSUB protocols as needed.
-		if subs, ok := subsNeedSUB[route]; ok && len(subs) > 0 {
-			route.sendRouteSubProtos(subs, false, nil)
+	// Now go over all accounts and invoke the update function defined above.
+	s.accounts.Range(func(_, v interface{}) bool {
+		acc := v.(*Account)
+		acc.mu.RLock()
+		accName, sl, poolIdx := acc.Name, acc.sl, acc.routePoolIdx
+		acc.mu.RUnlock()
+		if sl == nil {
+			return true
 		}
-		if unsubs, ok := subsNeedUNSUB[route]; ok && len(unsubs) > 0 {
-			route.sendRouteUnSubProtos(unsubs, false, nil)
+		var remotes []*client
+		for _, r := range routes {
+			r.mu.Lock()
+			if (poolIdx >= 0 && poolIdx == r.route.poolIdx) || (string(r.route.accName) == accName) {
+				remotes = append(remotes, r)
+			}
+			r.mu.Unlock()
 		}
-		route.mu.Unlock()
-	}
+		update(accName, sl, remotes)
+		return true
+	})
 }
 
 func (s *Server) reloadClusterPoolAndAccounts(co *clusterOption, opts *Options) {
diff --git a/server/route.go b/server/route.go
index f4123d2a..02161fba 100644
--- a/server/route.go
+++ b/server/route.go
@@ -927,39 +927,31 @@ func (s *Server) updateRemoteRoutePerms(c *client, info *Info) {
 
 	c.opts.Import = info.Import
 	c.opts.Export = info.Export
-
-	routeAcc, poolIdx, noPool := string(c.route.accName), c.route.poolIdx, c.route.noPool
 	c.mu.Unlock()
 
-	var (
-		_localSubs [4096]*subscription
-		_allSubs   [4096]*subscription
-		allSubs    = _allSubs[:0]
-	)
-
-	s.accounts.Range(func(_, v interface{}) bool {
-		acc := v.(*Account)
-		acc.mu.RLock()
-		accName, sl, accPoolIdx := acc.Name, acc.sl, acc.routePoolIdx
-		acc.mu.RUnlock()
-
-		// Do this only for accounts handled by this route
-		if (accPoolIdx >= 0 && accPoolIdx == poolIdx) || (routeAcc == accName) || noPool {
-			localSubs := _localSubs[:0]
-			sl.localSubs(&localSubs, false)
-			if len(localSubs) > 0 {
-				allSubs = append(allSubs, localSubs...)
-			}
-		}
-		return true
-	})
+	var acc *Account
+	var err error
+	if an := info.RouteAccount; an == _EMPTY_ {
+		acc = s.globalAccount()
+	} else if acc, err = s.LookupAccount(an); err != nil {
+		c.Errorf("Unable to lookup account %q to update remote route permissions: %v", an, err)
+		return
+	}
 
-	if len(allSubs) == 0 {
+	acc.mu.RLock()
+	sl := acc.sl
+	acc.mu.RUnlock()
+	if sl == nil {
 		return
 	}
+	var (
+		_localSubs [4096]*subscription
+		localSubs  = _localSubs[:0]
+	)
+	sl.localSubs(&localSubs, false)
 
 	c.mu.Lock()
-	c.sendRouteSubProtos(allSubs, false, func(sub *subscription) bool {
+	c.sendRouteSubProtos(localSubs, false, func(sub *subscription) bool {
 		subj := string(sub.subject)
 		// If the remote can now export but could not before, and this server can import this
 		// subject, then send SUB protocol.


Test patch:
diff --git a/server/reload_test.go b/server/reload_test.go
index d554bba0..66875088 100644
--- a/server/reload_test.go
+++ b/server/reload_test.go
@@ -4525,12 +4525,12 @@ func TestConfigReloadRouteImportPermissionsWithAccounts(t *testing.T) {
 		poolSize string
 		accounts string
 	}{
-		{"regular", "pool_size: -1", _EMPTY_},
+		{"regular", _EMPTY_, _EMPTY_},
 		{"pooling", "pool_size: 5", _EMPTY_},
 		{"per-account", _EMPTY_, "accounts: [\"A\"]"},
 		{"pool and per-account", "pool_size: 3", "accounts: [\"A\"]"},
 	} {
-		t.Run("import "+test.name, func(t *testing.T) {
+		t.Run(test.name, func(t *testing.T) {
 			confATemplate := `
 				server_name: "A"
 				port: -1
@@ -4604,7 +4604,6 @@ func TestConfigReloadRouteImportPermissionsWithAccounts(t *testing.T) {
 			natsFlush(t, ncA)
 
 			checkSubInterest(t, srvb, "A", "foo", 2*time.Second)
-			checkSubNoInterest(t, srvb, "A", "bar", 2*time.Second)
 
 			ncB := natsConnect(t, srvb.ClientURL(), nats.UserInfo("user1", "pwd"))
 			defer ncB.Close()
@@ -4614,7 +4613,7 @@ func TestConfigReloadRouteImportPermissionsWithAccounts(t *testing.T) {
 				if expected {
 					natsNexMsg(t, sub, time.Second)
 				} else {
-					if msg, err := sub.NextMsg(50 * time.Millisecond); err == nil {
+					if msg, err := sub.NextMsg(250 * time.Millisecond); err == nil {
 						t.Fatalf("Should not have gotten the message, got %s/%s", msg.Subject, msg.Data)
 					}
 				}
@@ -4635,7 +4634,6 @@ func TestConfigReloadRouteImportPermissionsWithAccounts(t *testing.T) {
 
 			checkClusterFormed(t, srva, srvb)
 
-			checkSubNoInterest(t, srvb, "A", "foo", 2*time.Second)
 			checkSubInterest(t, srvb, "A", "bar", 2*time.Second)
 
 			// Should not receive on foo
@@ -4655,158 +4653,6 @@ func TestConfigReloadRouteImportPermissionsWithAccounts(t *testing.T) {
 			checkClusterFormed(t, srva, srvb)
 
 			checkSubInterest(t, srvb, "A", "foo", 2*time.Second)
-			checkSubNoInterest(t, srvb, "A", "bar", 2*time.Second)
-
-			// Should receive on "foo"
-			natsPub(t, ncB, "foo", []byte("foo3"))
-			check(sub1Foo, true)
-			check(sub2Foo, true)
-			// But make sure there are no more than what we expect
-			check(sub1Foo, false)
-			check(sub2Foo, false)
-
-			// And now "bar" should fail
-			natsPub(t, ncB, "bar", []byte("bar3"))
-			check(sub1Bar, false)
-			check(sub2Bar, false)
-		})
-	}
-	// Check export now
-	for _, test := range []struct {
-		name     string
-		poolSize string
-		accounts string
-	}{
-		{"regular", "pool_size: -1", _EMPTY_},
-		{"pooling", "pool_size: 5", _EMPTY_},
-		{"per-account", _EMPTY_, "accounts: [\"A\"]"},
-		{"pool and per-account", "pool_size: 3", "accounts: [\"A\"]"},
-	} {
-		t.Run("export "+test.name, func(t *testing.T) {
-			confATemplate := `
-				server_name: "A"
-				port: -1
-				accounts {
-					A { users: [{user: "user1", password: "pwd"}] }
-					B { users: [{user: "user2", password: "pwd"}] }
-					C { users: [{user: "user3", password: "pwd"}] }
-					D { users: [{user: "user4", password: "pwd"}] }
-				}
-				cluster {
-					name: "local"
-					listen: 127.0.0.1:-1
-					permissions {
-						import {
-							allow: ">"
-						}
-						export {
-							allow: %s
-						}
-					}
-					%s
-					%s
-				}
-			`
-			confA := createConfFile(t, []byte(fmt.Sprintf(confATemplate, `"foo"`, test.poolSize, test.accounts)))
-			srva, optsA := RunServerWithConfig(confA)
-			defer srva.Shutdown()
-
-			confBTemplate := `
-				server_name: "B"
-				port: -1
-				accounts {
-					A { users: [{user: "user1", password: "pwd"}] }
-					B { users: [{user: "user2", password: "pwd"}] }
-					C { users: [{user: "user3", password: "pwd"}] }
-					D { users: [{user: "user4", password: "pwd"}] }
-				}
-				cluster {
-					listen: 127.0.0.1:-1
-					name: "local"
-					permissions {
-						import {
-							allow: ">"
-						}
-						export {
-							allow: %s
-						}
-					}
-					routes = [
-						"nats://127.0.0.1:%d"
-					]
-					%s
-					%s
-				}
-			`
-			confB := createConfFile(t, []byte(fmt.Sprintf(confBTemplate, `"foo"`, optsA.Cluster.Port, test.poolSize, test.accounts)))
-			srvb, _ := RunServerWithConfig(confB)
-			defer srvb.Shutdown()
-
-			checkClusterFormed(t, srva, srvb)
-
-			ncA := natsConnect(t, srva.ClientURL(), nats.UserInfo("user1", "pwd"))
-			defer ncA.Close()
-
-			sub1Foo := natsSubSync(t, ncA, "foo")
-			sub2Foo := natsSubSync(t, ncA, "foo")
-
-			sub1Bar := natsSubSync(t, ncA, "bar")
-			sub2Bar := natsSubSync(t, ncA, "bar")
-
-			natsFlush(t, ncA)
-
-			checkSubInterest(t, srvb, "A", "foo", 2*time.Second)
-			checkSubNoInterest(t, srvb, "A", "bar", 2*time.Second)
-
-			ncB := natsConnect(t, srvb.ClientURL(), nats.UserInfo("user1", "pwd"))
-			defer ncB.Close()
-
-			check := func(sub *nats.Subscription, expected bool) {
-				t.Helper()
-				if expected {
-					natsNexMsg(t, sub, time.Second)
-				} else {
-					if msg, err := sub.NextMsg(50 * time.Millisecond); err == nil {
-						t.Fatalf("Should not have gotten the message, got %s/%s", msg.Subject, msg.Data)
-					}
-				}
-			}
-
-			// Should receive on "foo"
-			natsPub(t, ncB, "foo", []byte("foo1"))
-			check(sub1Foo, true)
-			check(sub2Foo, true)
-
-			// But not on "bar"
-			natsPub(t, ncB, "bar", []byte("bar1"))
-			check(sub1Bar, false)
-			check(sub2Bar, false)
-
-			reloadUpdateConfig(t, srva, confA, fmt.Sprintf(confATemplate, `["foo", "bar"]`, test.poolSize, test.accounts))
-			reloadUpdateConfig(t, srvb, confB, fmt.Sprintf(confBTemplate, `["foo", "bar"]`, optsA.Cluster.Port, test.poolSize, test.accounts))
-
-			checkClusterFormed(t, srva, srvb)
-
-			checkSubInterest(t, srvb, "A", "foo", 2*time.Second)
-			checkSubInterest(t, srvb, "A", "bar", 2*time.Second)
-
-			// Should receive on foo and bar
-			natsPub(t, ncB, "foo", []byte("foo2"))
-			check(sub1Foo, true)
-			check(sub2Foo, true)
-
-			natsPub(t, ncB, "bar", []byte("bar2"))
-			check(sub1Bar, true)
-			check(sub2Bar, true)
-
-			// Remove "bar"
-			reloadUpdateConfig(t, srva, confA, fmt.Sprintf(confATemplate, `"foo"`, test.poolSize, test.accounts))
-			reloadUpdateConfig(t, srvb, confB, fmt.Sprintf(confBTemplate, `"foo"`, optsA.Cluster.Port, test.poolSize, test.accounts))
-
-			checkClusterFormed(t, srva, srvb)
-
-			checkSubInterest(t, srvb, "A", "foo", 2*time.Second)
-			checkSubNoInterest(t, srvb, "A", "bar", 2*time.Second)
 
 			// Should receive on "foo"
 			natsPub(t, ncB, "foo", []byte("foo3"))
diff --git a/server/routes_test.go b/server/routes_test.go
index 11dd380d..d60675d0 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -66,20 +66,6 @@ func checkSubInterest(t *testing.T, s *Server, accName, subject string, timeout
 	})
 }
 
-func checkSubNoInterest(t *testing.T, s *Server, accName, subject string, timeout time.Duration) {
-	t.Helper()
-	checkFor(t, timeout, 15*time.Millisecond, func() error {
-		acc, err := s.LookupAccount(accName)
-		if err != nil {
-			return fmt.Errorf("error looking up account %q: %v", accName, err)
-		}
-		if acc.SubscriptionInterest(subject) {
-			return fmt.Errorf("unexpected subscription interest for account %q on %q", accName, subject)
-		}
-		return nil
-	})
-}
-
 func TestRouteConfig(t *testing.T) {
 	opts, err := ProcessConfigFile("./configs/cluster.conf")
 	if err != nil {