Repository: nats-io/nats-server
Pull request #2479 (2021-08-30 22:50:36)

[FIXED] JetStream: possible deadlock due to lock inversion
The locking is jetStream->Server, not the otherway around. There
was few places where lock inversion could have caused deadlock.

Also, a change made recently to solve a deadlock was causing
a race that is demonstrated with TestJetStreamRaceOnRAFTCreate.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/jetstream.go b/server/jetstream.go
index a9d4b087..27b05fb3 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1307,6 +1307,7 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 	var stats JetStreamAccountStats
 	if jsa != nil {
 		js := jsa.js
+		js.mu.RLock()
 		jsa.mu.RLock()
 		stats.Memory = uint64(jsa.memTotal)
 		stats.Store = uint64(jsa.storeTotal)
@@ -1316,13 +1317,11 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 			Errors: jsa.apiErrors,
 		}
 		if cc := jsa.js.cluster; cc != nil {
-			js.mu.RLock()
 			sas := cc.streams[aname]
 			stats.Streams = len(sas)
 			for _, sa := range sas {
 				stats.Consumers += len(sa.consumers)
 			}
-			js.mu.RUnlock()
 		} else {
 			stats.Streams = len(jsa.streams)
 			for _, mset := range jsa.streams {
@@ -1331,6 +1330,7 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 		}
 		stats.Limits = jsa.limits
 		jsa.mu.RUnlock()
+		js.mu.RUnlock()
 	}
 	return stats
 }
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index ae968b63..848d4c62 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1187,6 +1187,7 @@ func (rg *raftGroup) setPreferred() {
 // createRaftGroup is called to spin up this raft group if needed.
 func (js *jetStream) createRaftGroup(rg *raftGroup, storage StorageType) error {
 	js.mu.Lock()
+	defer js.mu.Unlock()
 	s, cc := js.srv, js.cluster
 	if cc == nil || cc.meta == nil {
 		js.mu.Unlock()
@@ -1196,20 +1197,15 @@ func (js *jetStream) createRaftGroup(rg *raftGroup, storage StorageType) error {
 	// If this is a single peer raft group or we are not a member return.
 	if len(rg.Peers) <= 1 || !rg.isMember(cc.meta.ID()) {
 		// Nothing to do here.
-		js.mu.Unlock()
 		return nil
 	}
 
 	// Check if we already have this assigned.
-	// This server lock is not normal server lock, so ok to hold js lock here.
 	if node := s.lookupRaftNode(rg.Name); node != nil {
 		s.Debugf("JetStream cluster already has raft group %q assigned", rg.Name)
 		rg.node = node
-		js.mu.Unlock()
 		return nil
 	}
-	// Make sure to release lock here since below we will want the server lock.
-	js.mu.Unlock()
 
 	s.Debugf("JetStream cluster creating raft group:%+v", rg)
 
@@ -1251,10 +1247,7 @@ func (js *jetStream) createRaftGroup(rg *raftGroup, storage StorageType) error {
 		s.Debugf("Error creating raft group: %v", err)
 		return err
 	}
-	// js lock needs to be held here for assignment.
-	js.mu.Lock()
 	rg.node = n
-	js.mu.Unlock()
 
 	// See if we are preferred and should start campaign immediately.
 	if n.ID() == rg.Preferred {
diff --git a/server/monitor.go b/server/monitor.go
index fe87c7eb..40eb67da 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1259,6 +1259,21 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 	)
 }
 
+func (s *Server) updateJszVarz(js *jetStream, v *JetStreamVarz, doConfig bool) {
+	if doConfig {
+		js.mu.RLock()
+		// We want to snapshot the config since it will then be available outside
+		// of the js lock. So make a copy first, then point to this copy.
+		cfg := js.config
+		v.Config = &cfg
+		js.mu.RUnlock()
+	}
+	v.Stats = js.usageStats()
+	if mg := js.getMetaGroup(); mg != nil {
+		v.Meta = s.raftNodeToClusterInfo(mg)
+	}
+}
+
 // Varz returns a Varz struct containing the server information.
 func (s *Server) Varz(varzOpts *VarzOptions) (*Varz, error) {
 	var rss, vss int64
@@ -1268,11 +1283,15 @@ func (s *Server) Varz(varzOpts *VarzOptions) (*Varz, error) {
 	pse.ProcUsage(&pcpu, &rss, &vss)
 
 	s.mu.Lock()
+	js := s.js
 	// We need to create a new instance of Varz (with no reference
 	// whatsoever to anything stored in the server) since the user
 	// has access to the returned value.
 	v := s.createVarz(pcpu, rss)
 	s.mu.Unlock()
+	if js != nil {
+		s.updateJszVarz(js, &v.JetStream, true)
+	}
 
 	return v, nil
 }
@@ -1374,14 +1393,6 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 		}
 		varz.LeafNode.Remotes = rlna
 	}
-	if s.js != nil {
-		s.js.mu.RLock()
-		cfg := s.js.config
-		varz.JetStream = JetStreamVarz{
-			Config: &cfg,
-		}
-		s.js.mu.RUnlock()
-	}
 
 	// Finish setting it up with fields that can be updated during
 	// configuration reload and runtime.
@@ -1500,16 +1511,6 @@ func (s *Server) updateVarzRuntimeFields(v *Varz, forceUpdate bool, pcpu float64
 		}
 	}
 	gw.RUnlock()
-
-	if s.js != nil {
-		// FIXME(dlc) - We have lock inversion that needs to be fixed up properly.
-		s.mu.Unlock()
-		v.JetStream.Stats = s.js.usageStats()
-		if mg := s.js.getMetaGroup(); mg != nil {
-			v.JetStream.Meta = s.raftNodeToClusterInfo(mg)
-		}
-		s.mu.Lock()
-	}
 }
 
 // HandleVarz will process HTTP requests for server information.
@@ -1531,13 +1532,32 @@ func (s *Server) HandleVarz(w http.ResponseWriter, r *http.Request) {
 
 	// Use server lock to create/update the server's varz object.
 	s.mu.Lock()
+	var created bool
+	js := s.js
 	s.httpReqStats[VarzPath]++
 	if s.varz == nil {
 		s.varz = s.createVarz(pcpu, rss)
+		created = true
 	} else {
 		s.updateVarzRuntimeFields(s.varz, false, pcpu, rss)
 	}
 	s.mu.Unlock()
+	// Since locking is jetStream -> Server, need to update jetstream
+	// varz outside of server lock.
+	if js != nil {
+		var v JetStreamVarz
+		// Work on stack variable
+		s.updateJszVarz(js, &v, created)
+		// Now update server's varz
+		s.mu.Lock()
+		sv := &s.varz.JetStream
+		if created {
+			sv.Config = v.Config
+		}
+		sv.Stats = v.Stats
+		sv.Meta = v.Meta
+		s.mu.Unlock()
+	}
 
 	// Do the marshaling outside of server lock, but under varzMu lock.
 	b, err := json.MarshalIndent(s.varz, "", "  ")


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 88c6ec16..02e07790 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -8002,6 +8002,82 @@ func TestJetStreamClusterConsumerLastActiveReporting(t *testing.T) {
 	checkTimeDiff(ci.AckFloor.Last, nci.AckFloor.Last)
 }
 
+func TestJetStreamRaceOnRAFTCreate(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	srv := c.servers[0]
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	}); err != nil {
+		t.Fatalf("Error creating stream: %v", err)
+	}
+
+	size := 10
+	wg := sync.WaitGroup{}
+	wg.Add(size)
+	for i := 0; i < size; i++ {
+		go func() {
+			defer wg.Done()
+			js.PullSubscribe("foo", "shared")
+		}()
+	}
+	wg.Wait()
+}
+
+func TestJetStreamDeadlockOnVarz(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	srv := c.servers[0]
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	size := 10
+	wg := sync.WaitGroup{}
+	wg.Add(size)
+	ch := make(chan struct{})
+	for i := 0; i < size; i++ {
+		go func(i int) {
+			defer wg.Done()
+			<-ch
+			js.AddStream(&nats.StreamConfig{
+				Name:     fmt.Sprintf("TEST%d", i),
+				Subjects: []string{"foo"},
+				Replicas: 3,
+			})
+		}(i)
+	}
+
+	close(ch)
+	for i := 0; i < 10; i++ {
+		srv.Varz(nil)
+		time.Sleep(time.Millisecond)
+	}
+	wg.Wait()
+}
+
 // Support functions
 
 // Used to setup superclusters for tests.