Merge #105642

105642: kvserver: de-flake TestMergeQueueSeesLearnerOrJointConfig r=irfansharif a=irfansharif

Fixes #105381. It was possible for us to receive client-side store IDs concurrently with the server-side stopper being quiesced, at which point we read the variable storing the client-side store IDs without a mutex (which triggered the data race detector). The data race was benign, but instead of adding a synchronization primitive, we recognize that there was no reason for the store ID handling to happen on separate threads.

Release note: None

Co-authored-by: irfan sharif <[email protected]>

pkg/kv/kvserver/raft_transport.go

@@ -448,25 +448,25 @@ func (t *RaftTransport) RaftMessageBatch(stream MultiRaft_RaftMessageBatchServer
 	taskCtx = t.AnnotateCtx(taskCtx)
 	defer cancel()
 
-	var storeIDs []roachpb.StoreID
-	defer func() {
-		ctx := t.AnnotateCtx(context.Background())
-		t.kvflowControl.mu.Lock()
-		t.kvflowControl.mu.connectionTracker.markStoresDisconnected(storeIDs)
-		t.kvflowControl.mu.Unlock()
-		t.kvflowControl.disconnectListener.OnRaftTransportDisconnected(ctx, storeIDs...)
-		if fn := t.knobs.OnServerStreamDisconnected; fn != nil {
-			fn()
-		}
-	}()
-
 	if err := t.stopper.RunAsyncTaskEx(
 		taskCtx,
 		stop.TaskOpts{
 			TaskName: "storage.RaftTransport: processing batch",
 			SpanOpt:  stop.ChildSpan,
 		}, func(ctx context.Context) {
 			errCh <- func() error {
+				var storeIDs []roachpb.StoreID
+				defer func() {
+					ctx := t.AnnotateCtx(context.Background())
+					t.kvflowControl.mu.Lock()
+					t.kvflowControl.mu.connectionTracker.markStoresDisconnected(storeIDs)
+					t.kvflowControl.mu.Unlock()
+					t.kvflowControl.disconnectListener.OnRaftTransportDisconnected(ctx, storeIDs...)
+					if fn := t.knobs.OnServerStreamDisconnected; fn != nil {
+						fn()
+					}
+				}()
+
 				stream := &lockedRaftMessageResponseStream{wrapped: stream}
 				for {
 					batch, err := stream.Recv()