[FIXED] Dedupe map cleanup race on leader change

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

locksordering.txt

@@ -41,7 +41,7 @@ locking the stream lock afterward would violate locking order.
 
     stream -> clMu
     stream -> batchMu -> clMu
-    stream -> ddMu
+    stream -> clMu -> ddMu
 
 The "mset.batches.mu" lock protects the batching state without needing to hold the stream lock.
 If "clMu" is used to commit a batch, it should only be acquired while already holding the batch lock.

server/jetstream_cluster.go

@@ -4526,6 +4526,10 @@ func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
 		return
 	}
 
+	// Acquire clMu before ddMu so any inflight proposals finish first, and we can
+	// clean up if they added new dedupe IDs.
+	mset.clMu.Lock()
+
 	// Clear inflight dedupe IDs, where seq=0.
 	mset.ddMu.Lock()
 	var removed int
@@ -4548,7 +4552,6 @@ func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
 	}
 	mset.ddMu.Unlock()
 
-	mset.clMu.Lock()
 	// Clear inflight if we have it.
 	mset.inflight = nil
 	mset.inflightTransform = nil
@@ -4557,6 +4560,12 @@ func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
 	// Clear expected per subject state.
 	mset.expectedPerSubjectSequence = nil
 	mset.expectedPerSubjectInProcess = nil
+
+	// Clear clseq on every leader transition. recalculateClusteredSeq
+	// repopulates it on the next proposal.
+	if mset.clseq > 0 {
+		mset.clseq = 0
+	}
 	mset.clMu.Unlock()
 
 	js.mu.RLock()
@@ -4578,14 +4587,6 @@ func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
 		}
 	}
 
-	// Clear clseq on every leader transition. recalculateClusteredSeq
-	// repopulates it on the next proposal.
-	mset.clMu.Lock()
-	if mset.clseq > 0 {
-		mset.clseq = 0
-	}
-	mset.clMu.Unlock()
-
 	// Tell stream to switch leader status.
 	mset.setLeader(isLeader)
 

server/jetstream_cluster_4_test.go

@@ -2548,6 +2548,69 @@ func TestJetStreamClusterPubAckSequenceDupeResetAfterLeaderChange(t *testing.T)
 	require_NoError(t, err)
 }
 
+func TestJetStreamClusterStreamLeaderChangeDedupeCleanupRace(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,
+		Duplicates: 2 * time.Minute,
+	})
+	require_NoError(t, err)
+
+	sl := c.streamLeader(globalAccountName, "TEST")
+	require_NotNil(t, sl)
+	mset, err := sl.globalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	sjs := sl.getJetStream()
+	require_NotNil(t, sjs)
+
+	// Simulate an inflight proposal: holds clMu and is about to plant a seq=0 placeholder.
+	mset.clMu.Lock()
+
+	// In the meantime, we become leader again. processStreamLeaderChange must block on
+	// clMu so the placeholder is in ddmap by the time the cleanup runs.
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		sjs.processStreamLeaderChange(mset, true)
+	}()
+
+	// Give the goroutine time to reach the clMu acquisition.
+	time.Sleep(200 * time.Millisecond)
+	select {
+	case <-done:
+		t.Fatal("processStreamLeaderChange should be blocked on clMu")
+	default:
+	}
+
+	// Plant the seq=0 placeholder, simulating a proposal's diff.commit step
+	// happening while the leader change is in flight.
+	mset.ddMu.Lock()
+	mset.storeMsgIdLocked(&ddentry{"msgId", 0, time.Now().UnixNano()})
+	mset.ddMu.Unlock()
+
+	// Release clMu so processStreamLeaderChange can proceed with cleanup.
+	mset.clMu.Unlock()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("processStreamLeaderChange did not complete after clMu released")
+	}
+
+	// The placeholder must have been cleaned up.
+	mset.ddMu.Lock()
+	defer mset.ddMu.Unlock()
+	require_Len(t, len(mset.ddmap), 0)
+	require_Len(t, len(mset.ddarr), 0)
+}
+
 func TestJetStreamClusterConsumeWithStartSequence(t *testing.T) {
 
 	const (