[FIXED] Route interest subscription race

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

server/accounts.go

@@ -66,7 +66,7 @@ type Account struct {
 	claimJWT     string
 	updated      time.Time
 	mu           sync.RWMutex
-	sqmu         sync.Mutex
+	smu          sync.Mutex // serializes route interest updates
 	sl           *Sublist
 	ic           *client
 	sq           *sendq
@@ -80,7 +80,7 @@ type Account struct {
 	nrleafs      int32
 	clients      map[*client]struct{}
 	rm           map[string]int32
-	lqws         map[string]int32
+	lws          map[string]int32 // per key, last rm[key] sent to routes; used to dedup sends
 	usersRevoked map[string]int64
 	mappings     []*mapping
 	hasMapped    atomic.Bool

server/route.go

@@ -2539,21 +2539,21 @@ func (s *Server) updateRouteSubscriptionMap(acc *Account, sub *subscription, del
 		// queue subscriptions updates (sub/unsub).
 		// See https://github.com/nats-io/nats-server/pull/1126 for more details.
 		if isq {
-			acc.sqmu.Lock()
+			acc.smu.Lock()
 		}
 		acc.mu.Lock()
 	}
 	accUnlock := func() {
 		acc.mu.Unlock()
 		if isq {
-			acc.sqmu.Unlock()
+			acc.smu.Unlock()
 		}
 	}
 
 	accLock()
 
 	// This is non-nil when we know we are in cluster mode.
-	rm, lqws := acc.rm, acc.lqws
+	rm, lws := acc.rm, acc.lws
 	if rm == nil {
 		accUnlock()
 		return
@@ -2573,9 +2573,7 @@ func (s *Server) updateRouteSubscriptionMap(acc *Account, sub *subscription, del
 		n += delta
 		if n <= 0 {
 			delete(rm, key)
-			if isq {
-				delete(lqws, key)
-			}
+			delete(lws, key)
 			update = true // Update for deleting (N->0)
 		} else {
 			rm[key] = n
@@ -2650,29 +2648,29 @@ func (s *Server) updateRouteSubscriptionMap(acc *Account, sub *subscription, del
 	trace := atomic.LoadInt32(&s.logging.trace) == 1
 	s.mu.RUnlock()
 
-	// If we are a queue subscriber we need to make sure our updates are serialized from
-	// potential multiple connections. We want to make sure that the order above is preserved
-	// here but not necessarily all updates need to be sent. We need to block and recheck the
-	// n count with the lock held through sending here. We will suppress duplicate sends of same qw.
-	if isq {
-		// However, we can't hold the acc.mu lock since we allow client.mu.Lock -> acc.mu.Lock
-		// but not the opposite. So use a dedicated lock while holding the route's lock.
-		acc.sqmu.Lock()
-		defer acc.sqmu.Unlock()
+	// We need to make sure our updates are serialized from potential multiple connections. We want
+	// to make sure that the order above is preserved here but not necessarily all updates need to
+	// be sent. We need to block and recheck the n count with the lock held through sending here.
+	//
+	// However, we can't hold the acc.mu lock since we allow client.mu.Lock -> acc.mu.Lock
+	// but not the opposite. So use a dedicated lock while holding the route's lock.
+	acc.smu.Lock()
+	defer acc.smu.Unlock()
 
-		acc.mu.Lock()
-		n = rm[key]
+	acc.mu.Lock()
+	n = rm[key]
+	if isq {
 		sub.qw = n
-		// Check the last sent weight here. If same, then someone
-		// beat us to it and we can just return here. Otherwise update
-		if ls, ok := lqws[key]; ok && ls == n {
-			acc.mu.Unlock()
-			return
-		} else if n > 0 {
-			lqws[key] = n
-		}
-		acc.mu.Unlock()
 	}
+	// Check the last sent value here. If same, then someone beat us to it and
+	// we can just return here. Otherwise update.
+	if ls, ok := lws[key]; ok && ls == n {
+		acc.mu.Unlock()
+		return
+	} else if n > 0 {
+		lws[key] = n
+	}
+	acc.mu.Unlock()
 
 	// Snapshot into array
 	subs := []*subscription{sub}

server/routes_test.go

@@ -1638,14 +1638,14 @@ func TestClusterQueueGroupWeightTrackingLeak(t *testing.T) {
 		key := keyFromSubWithOrigin(&sub)
 		checkFor(t, time.Second, 15*time.Millisecond, func() error {
 			acc.mu.RLock()
-			v, ok := acc.lqws[key]
+			v, ok := acc.lws[key]
 			acc.mu.RUnlock()
 			if present {
 				if !ok {
 					return fmt.Errorf("the key is not present")
 				}
 				if v != expected {
-					return fmt.Errorf("lqws doest not contain expected value of %v: %v", expected, v)
+					return fmt.Errorf("lws does not contain expected value of %v: %v", expected, v)
 				}
 			} else if ok {
 				return fmt.Errorf("the key is present with value %v and should not be", v)
@@ -5440,6 +5440,114 @@ func TestRoutePoolFirstPongBlocksChain(t *testing.T) {
 	}
 }
 
+// https://github.com/nats-io/nats-server/issues/8233
+func TestRouteSubUnsubRaceLosesRemoteInterest(t *testing.T) {
+	oa := DefaultOptions()
+	sa := RunServer(oa)
+	defer sa.Shutdown()
+
+	ob := DefaultOptions()
+	ob.Routes = RoutesFromStr(fmt.Sprintf("nats://127.0.0.1:%d", oa.Cluster.Port))
+	sb := RunServer(ob)
+	defer sb.Shutdown()
+
+	checkClusterFormed(t, sa, sb)
+
+	// A client on node B, used to confirm the user-visible symptom: a request to
+	// the subject gets "no responders" even though a live responder exists on A.
+	ncReq := natsConnect(t, sb.ClientURL())
+	defer ncReq.Close()
+
+	// Many subjects raced at once so they all contend the same global-account
+	// acc.mu and the single pinned A->B route.mu.
+	const (
+		conns = 100
+		waves = 10
+	)
+
+	// Connection pools, reused across waves. ncMinus owns the original subscription
+	// that gets removed, ncPlus adds the new responder that must survive. They are
+	// distinct connections so their operations run on different readLoop goroutines
+	// on node A and can race.
+	ncMinus := make([]*nats.Conn, conns)
+	ncPlus := make([]*nats.Conn, conns)
+	for i := 0; i < conns; i++ {
+		ncMinus[i] = natsConnect(t, sa.ClientURL())
+		defer ncMinus[i].Close()
+		ncPlus[i] = natsConnect(t, sa.ClientURL())
+		defer ncPlus[i].Close()
+	}
+
+	accB := sb.globalAccount()
+	for w := range waves {
+		subjects := make([]string, conns)
+		minusSubs := make([]*nats.Subscription, conns)
+		for i := range conns {
+			subjects[i] = fmt.Sprintf("foo.%d.%d", w, i)
+			// Original interest, mirrored to B via RS+. This is the "live"
+			// subscription on B that makes the reordered RS+ a no-op.
+			minusSubs[i] = natsSubSync(t, ncMinus[i], subjects[i])
+		}
+		for i := range conns {
+			natsFlush(t, ncMinus[i])
+		}
+		// Make sure B sees all of the original interest before we start, so its
+		// route already tracks each key.
+		for i := range conns {
+			checkSubInterest(t, sb, globalAccountName, subjects[i], 5*time.Second)
+		}
+
+		// Fire the wave: for each subject, simultaneously remove the original
+		// interest (-1) and add a replacement responder (+1) on node A.
+		start := make(chan struct{})
+		var wg sync.WaitGroup
+		wg.Add(2 * conns)
+		for i := range conns {
+			go func() {
+				defer wg.Done()
+				<-start
+				minusSubs[i].Unsubscribe()
+				ncMinus[i].Flush()
+			}()
+			go func() {
+				defer wg.Done()
+				<-start
+				ncPlus[i].Subscribe(subjects[i], func(m *nats.Msg) {
+					m.Respond([]byte("ok"))
+				})
+				ncPlus[i].Flush()
+			}()
+		}
+		close(start)
+		wg.Wait()
+
+		// Barrier: subscribe a marker on the same (pinned) route and wait for B
+		// to observe it. Because all of the wave's RS+/RS- protos were enqueued
+		// on that same route connection before this one, FIFO ordering
+		// guarantees that once B has the marker it has processed every proto
+		// from the wave.
+		marker := fmt.Sprintf("marker.%d", w)
+		natsSubSync(t, ncPlus[0], marker)
+		natsFlush(t, ncPlus[0])
+		checkSubInterest(t, sb, globalAccountName, marker, 5*time.Second)
+
+		// Every subject still has a live local responder on A (the ncPlus sub),
+		// so B must still have interest. If the race fired, B silently lost it.
+		for i := range conns {
+			if accB.SubscriptionInterest(subjects[i]) {
+				continue
+			}
+			// Confirm the exact reported symptom: a request from a client on B
+			// gets "no responders" although A has a live responder.
+			_, rerr := ncReq.Request(subjects[i], []byte("ping"), time.Second)
+			t.Fatalf("wave %d: node B lost interest in %q while node A still has a "+
+				"live local responder; request from B returned %v (want a reply). "+
+				"RS+/RS- were reordered on the A->B route by the "+
+				"updateRouteSubscriptionMap race", w, subjects[i], rerr)
+		}
+	}
+}
+
 // Benchmarks for message arg processing functions to measure heap allocations.
 // These functions parse incoming protocol messages and split arguments.
 

server/server.go

@@ -1968,7 +1968,7 @@ func (s *Server) registerAccountNoLock(acc *Account) *Account {
 	// TODO(dlc)- Double check that we need this for GWs.
 	if acc.rm == nil && s.opts != nil && s.shouldTrackSubscriptions() {
 		acc.rm = make(map[string]int32)
-		acc.lqws = make(map[string]int32)
+		acc.lws = make(map[string]int32)
 	}
 	acc.srv = s
 	acc.updated = time.Now()