[FIXED] Duplicate shadow durable queue sub due to race

server/server.go

@@ -907,9 +907,7 @@ func (ss *subStore) Store(sub *subState) error {
 		return err
 	}
 
-	ss.Lock()
 	ss.updateState(sub)
-	ss.Unlock()
 
 	return nil
 }
@@ -938,6 +936,9 @@ func (ss *subStore) updateState(sub *subState) {
 			// maybe due to upgrades from much older releases that had bugs?).
 			// So don't panic and use as the shadow the one with the highest LastSent
 			// value.
+			if qs.shadow != nil {
+				ss.stan.log.Warnf("Duplicate shadow durable queue consumer (subid=%v) for group %q", sub.ID, sub.QGroup)
+			}
 			if qs.shadow == nil || sub.LastSent > qs.lastSent {
 				qs.shadow = sub
 			}
@@ -2361,17 +2362,21 @@ func (s *StanServer) processRecoveredChannels(channels map[string]*stores.Recove
 			return nil, err
 		}
 		if !s.isClustered {
+			ss := channel.ss
+			ss.Lock()
 			// Get the recovered subscriptions for this channel.
 			for _, recSub := range recoveredChannel.Subscriptions {
 				sub := s.recoverOneSub(channel, recSub.Sub, recSub.Pending, nil)
 				if sub != nil {
 					// Subscribe to subscription ACKs
 					if err := sub.startAckSub(s.nca, s.processAckMsg); err != nil {
+						ss.Unlock()
 						return nil, err
 					}
 					allSubs = append(allSubs, sub)
 				}
 			}
+			ss.Unlock()
 			// Now that we have recovered possible subscriptions for this channel,
 			// check if we should start the delete timer.
 			if channel.activity != nil {
@@ -3013,17 +3018,17 @@ func (s *StanServer) checkClientHealth(clientID string) {
 			// close the client (connection). This locks the
 			// client object internally so unlock here.
 			client.Unlock()
-			// If clustered, thread operations through Raft.
-			if s.isClustered {
-				s.barrier(func() {
+			s.barrier(func() {
+				// If clustered, thread operations through Raft.
+				if s.isClustered {
 					if err := s.replicateConnClose(&pb.CloseRequest{ClientID: clientID}); err != nil {
 						s.log.Errorf("[Client:%s] Failed to replicate disconnect on heartbeat expiration: %v",
 							clientID, err)
 					}
-				})
-			} else {
-				s.closeClient(clientID)
-			}
+				} else {
+					s.closeClient(clientID)
+				}
+			})
 			return
 		}
 	} else {
@@ -4605,15 +4610,13 @@ func (s *StanServer) updateDurable(ss *subStore, sub *subState, clientID string)
 	if err != nil {
 		return err
 	}
-	ss.Lock()
 	// Do this only for durable subscribers (not durable queue subscribers).
 	if sub.isDurableSubscriber() {
 		// Add back into plain subscribers
 		ss.psubs = append(ss.psubs, sub)
 	}
 	// And in ackInbox lookup map.
 	ss.acks[sub.AckInbox] = sub
-	ss.Unlock()
 
 	return nil
 }
@@ -4633,6 +4636,9 @@ func (s *StanServer) processSub(c *channel, sr *pb.SubscriptionRequest, ackInbox
 		sub *subState
 		ss  = c.ss
 	)
+
+	ss.Lock()
+
 	// Will be true for durable queue subscribers and durable subscribers alike.
 	isDurable := false
 	// Will be set to false for en existing durable subscriber or existing
@@ -4646,6 +4652,7 @@ func (s *StanServer) processSub(c *channel, sr *pb.SubscriptionRequest, ackInbox
 			if strings.Contains(sr.DurableName, ":") {
 				s.log.Errorf("[Client:%s] Invalid DurableName (%q) for queue subscriber from %s",
 					sr.ClientID, sr.DurableName, sr.Subject)
+				ss.Unlock()
 				return nil, ErrInvalidDurName
 			}
 			isDurable = true
@@ -4656,7 +4663,6 @@ func (s *StanServer) processSub(c *channel, sr *pb.SubscriptionRequest, ackInbox
 		}
 		// Lookup for an existing group. Only interested in situation where
 		// the group exist, but is empty and had a shadow subscriber.
-		ss.RLock()
 		qs := ss.qsubs[sr.QGroup]
 		if qs != nil {
 			qs.Lock()
@@ -4668,15 +4674,15 @@ func (s *StanServer) processSub(c *channel, sr *pb.SubscriptionRequest, ackInbox
 			qs.Unlock()
 			setStartPos = false
 		}
-		ss.RUnlock()
 	} else if sr.DurableName != "" {
 		// Check for DurableSubscriber status
-		if sub = ss.LookupByDurable(durableKey(sr)); sub != nil {
+		if sub = ss.durables[durableKey(sr)]; sub != nil {
 			sub.RLock()
 			clientID := sub.ClientID
 			sub.RUnlock()
 			if clientID != "" {
 				s.log.Errorf("[Client:%s] Duplicate durable subscription registration", sr.ClientID)
+				ss.Unlock()
 				return nil, ErrDupDurable
 			}
 			setStartPos = false
@@ -4766,14 +4772,13 @@ func (s *StanServer) processSub(c *channel, sr *pb.SubscriptionRequest, ackInbox
 			sub.ID = subID
 			err = s.addSubscription(ss, sub)
 			if err == nil && subID > 0 {
-				ss.Lock()
 				if subID >= c.nextSubID {
 					c.nextSubID = subID + 1
 				}
-				ss.Unlock()
 			}
 		}
 	}
+	ss.Unlock()
 	if err == nil && (!s.isClustered || s.isLeader()) {
 		err = sub.startAckSub(s.nca, s.processAckMsg)
 		if err == nil {

server/server_queue_test.go

@@ -1161,3 +1161,111 @@ func TestQueueNoRedeliveryDuringSubClose(t *testing.T) {
 	default:
 	}
 }
+
+func TestPersistentStoreDurableQueueSubRaceBetweenCreateAndClose(t *testing.T) {
+	cleanupDatastore(t)
+	defer cleanupDatastore(t)
+
+	opts := getTestDefaultOptsForPersistentStore()
+	s := runServerWithOpts(t, opts, nil)
+	defer shutdownRestartedServerOnTestExit(&s)
+
+	sc := NewDefaultConnection(t)
+	defer sc.Close()
+
+	qsub1, err := sc.QueueSubscribe("foo", "bar", func(_ *stan.Msg) {}, stan.DurableName("dur"))
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+
+	cs := channelsGet(t, s.channels, "foo")
+	ss := &mockedSubStore{SubStore: cs.store.Subs}
+	cs.store.Subs = ss
+
+	// Will make the store create sub pause so that the addition of the new
+	// queue sub in the go routine will be delayed.
+	ss.Lock()
+	ss.ch = make(chan bool, 1)
+	ss.Unlock()
+	closeErrCh := make(chan error)
+	createErrCh := make(chan error)
+
+	sc2, err := stan.Connect(clusterName, clientName+"2")
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer sc2.Close()
+
+	// Since close operation uses a barrier that ensures that
+	// other operations complete, we need to start processing
+	// of qsub1.Close() start before we start the new queue sub,
+	// but ensure that it stops at one point. We will use the
+	// closeMu mutex to artificially block it.
+	s.closeMu.Lock()
+	go func() {
+		closeErrCh <- qsub1.Close()
+	}()
+	// Let the Close() proceed a bit...
+	time.Sleep(100 * time.Millisecond)
+	go func() {
+		_, err := sc2.QueueSubscribe("foo", "bar", func(_ *stan.Msg) {}, stan.DurableName("dur"))
+		createErrCh <- err
+	}()
+
+	// Let the QueueSubscribe() proceed a bit..
+	time.Sleep(100 * time.Millisecond)
+
+	// Now release the qsub1 Close()
+	s.closeMu.Unlock()
+
+	// Let the Close() complete, but we have to release the CreateSub()
+	// blockage from a different go routine otherwise with the fix,
+	// the Close() would still be blocked.
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		// Now release the addition to the subStore of the second queue sub
+		ss.ch <- true
+	}()
+
+	// Wait for the close to complete
+	if err := <-closeErrCh; err != nil {
+		t.Fatalf("Error on close: %v", err)
+	}
+	// Wait for create to complete
+	if err := <-createErrCh; err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+
+	// Now close 2nd queue sub (connection close will do that for us)
+	sc2.Close()
+	// We don't need that connection either.
+	sc.Close()
+
+	// Shutdown and restart the server and make sure that we did not
+	// introduce a 2nd shadow subscription.
+	s.Shutdown()
+	l := &captureWarnLogger{}
+	opts.CustomLogger = l
+	s = runServerWithOpts(t, opts, nil)
+
+	cs = channelsGet(t, s.channels, "foo")
+	qs := cs.ss.qsubs["dur:bar"]
+	qs.RLock()
+	hasShadow := qs.shadow != nil
+	qs.RUnlock()
+	if !hasShadow {
+		t.Fatalf("Should have a shadow subscription")
+	}
+	var hasDuplicate bool
+	l.Lock()
+	for _, w := range l.warnings {
+		if strings.Contains(w, "Duplicate shadow durable") {
+			hasDuplicate = true
+			break
+		}
+	}
+	l.Unlock()
+	if hasDuplicate {
+		t.Fatalf("Duplicate shadow subscription found!")
+	}
+}

server/server_storefailures_test.go

@@ -41,6 +41,7 @@ type mockedSubStore struct {
 	sync.RWMutex
 	fail          bool
 	failFlushOnce bool
+	ch            chan bool
 }
 
 func (ms *mockedStore) CreateChannel(name string) (*stores.Channel, error) {
@@ -253,7 +254,12 @@ func TestMsgLookupFailures(t *testing.T) {
 func (ss *mockedSubStore) CreateSub(sub *spb.SubState) error {
 	ss.RLock()
 	fail := ss.fail
+	ch := ss.ch
 	ss.RUnlock()
+	if ch != nil {
+		// Wait for notification that we can continue
+		<-ch
+	}
 	if fail {
 		return fmt.Errorf("On purpose")
 	}

server/snapshot.go

@@ -429,8 +429,8 @@ func (r *raftFSM) restoreChannelsFromSnapshot(serverSnap *spb.RaftSnapshot, inNe
 		c.lSeqChecked = false
 
 		for _, ss := range sc.Subscriptions {
-			s.recoverOneSub(c, ss.State, nil, ss.AcksPending)
 			c.ss.Lock()
+			s.recoverOneSub(c, ss.State, nil, ss.AcksPending)
 			if ss.State.ID >= c.nextSubID {
 				c.nextSubID = ss.State.ID + 1
 			}