[FIXED] Deleted raft node's stream is revived (#7668)

Related to #7025

If the Raft node of the stream was deleted while the stream was stalled
on upper-layer catchup, it would wrongfully revive the stream. We now
mark the Raft node as deleted so we can ensure the stream monitor
goroutine quits and doesn't get revived.

Signed-off-by: Maurice van Veen <[email protected]>

Author: neilalexander

server/jetstream_cluster.go

@@ -3142,20 +3142,20 @@ func (mset *stream) resetClusteredState(err error) bool {
 	stype, tierName, replicas := mset.cfg.Storage, mset.tier, mset.cfg.Replicas
 	mset.mu.RUnlock()
 
-	assert.Unreachable("Reset clustered state", map[string]any{
-		"stream":  name,
-		"account": acc.Name,
-		"err":     err,
-	})
-
 	// The stream might already be deleted and not assigned to us anymore.
 	// In any case, don't revive the stream if it's already closed.
-	if mset.closed.Load() {
+	if mset.closed.Load() || (node != nil && node.IsDeleted()) {
 		s.Warnf("Will not reset stream '%s > %s', stream is closed", acc, mset.name())
 		// Explicitly returning true here, we want the outside to break out of the monitoring loop as well.
 		return true
 	}
 
+	assert.Unreachable("Reset clustered state", map[string]any{
+		"stream":  name,
+		"account": acc.Name,
+		"err":     err,
+	})
+
 	// Stepdown regardless if we are the leader here.
 	if node != nil {
 		node.StepDown()

server/jetstream_cluster_3_test.go

@@ -6630,3 +6630,75 @@ func TestJetStreamClusterStreamDesyncDuringSnapshot(t *testing.T) {
 	t.Run("Reset", func(t *testing.T) { test(t, KindReset) })
 	t.Run("Truncate", func(t *testing.T) { test(t, KindTruncate) })
 }
+
+func TestJetStreamClusterDeletedNodeDoesNotReviveStreamAfterCatchup(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+		Storage:  nats.FileStorage,
+	})
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+	checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+		return checkState(t, c, globalAccountName, "TEST")
+	})
+
+	rs := c.randomNonStreamLeader(globalAccountName, "TEST")
+	for _, s := range c.servers {
+		if s == rs {
+			continue
+		}
+		s.Shutdown()
+		s.WaitForShutdown()
+	}
+
+	mset, err := rs.globalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	snap := mset.stateSnapshot()
+
+	// Reset the entire store so we can catchup based on the above snapshot.
+	fs := mset.store.(*fileStore)
+	require_NoError(t, fs.reset())
+
+	// Mark the node as leaderless, and get the upper-layer to start a catchup from a snapshot.
+	node := mset.raftNode()
+	node.(*raft).hasleader.Store(false)
+	node.ApplyQ().push(newCommittedEntry(10, []*Entry{{EntrySnapshot, snap}}))
+
+	// Since the node is leaderless, it will retry after some time. We wait a little here to ensure
+	// it's waiting there as well, and then we delete the node outright.
+	time.Sleep(time.Second)
+	node.Delete()
+
+	// The stream's goroutine should eventually be stopped. This will fail if the stream is revived.
+	var retries int
+	checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
+		mset, err = rs.globalAccount().lookupStream("TEST")
+		if err != nil {
+			retries = 0
+			return err
+		}
+		if mset.isMonitorRunning() {
+			retries = 0
+			return errors.New("monitor still running")
+		}
+		if state := mset.raftNode().State(); state != Closed {
+			retries = 0
+			return errors.New("node not closed")
+		}
+		retries++
+		if retries < 3 {
+			return errors.New("still confirming stable state")
+		}
+		return nil
+	})
+}

server/raft.go

@@ -83,6 +83,7 @@ type RaftNode interface {
 	Stop()
 	WaitForStop()
 	Delete()
+	IsDeleted() bool
 	RecreateInternalSubs() error
 	IsSystemAccount() bool
 	GetTrafficAccountName() string
@@ -231,6 +232,7 @@ type raft struct {
 	initializing bool // The node is new, and "empty log" checks can be temporarily relaxed.
 	scaleUp      bool // The node is part of a scale up, puts us in observer mode until the log contains data.
 	membChanging bool // There is a membership change proposal in progress
+	deleted      bool // If the node was deleted.
 }
 
 type proposedEntry struct {
@@ -1917,13 +1919,20 @@ func (n *raft) Delete() {
 	n.Lock()
 	defer n.Unlock()
 
+	n.deleted = true
 	if wal := n.wal; wal != nil {
 		wal.Delete(false)
 	}
 	os.RemoveAll(n.sd)
 	n.debug("Deleted")
 }
 
+func (n *raft) IsDeleted() bool {
+	n.RLock()
+	defer n.RUnlock()
+	return n.deleted
+}
+
 func (n *raft) shutdown() {
 	// First call to Stop or Delete should close the quit chan
 	// to notify the runAs goroutines to stop what they're doing.