Fix data race on Checksumming field in TGroupSessions (#26096) (#26231)

Author: serbel324

ydb/core/blobstorage/dsproxy/group_sessions.cpp

@@ -85,7 +85,7 @@ TGroupSessions::TGroupSessions(const TIntrusivePtr<TBlobStorageGroupInfo>& info,
             auto& q = stateVDisk.Queues.GetQueue(queueId);
             q.ActorId = queue;
             q.FlowRecord = std::move(flowRecord);
-            q.ExtraBlockChecksSupport.reset();
+            q.ExtraBlockChecksSupport.store(false);
         }
     }
 }
@@ -128,17 +128,17 @@ void TGroupSessions::QueueConnectUpdate(ui32 orderNumber, NKikimrBlobStorage::EV
 
     if (connected) {
         ConnectedQueuesMask[orderNumber] |= 1 << queueId;
-        q.ExtraBlockChecksSupport = extraGroupChecksSupport;
-        q.Checksumming = checksumming;
+        q.ExtraBlockChecksSupport.store(extraGroupChecksSupport);
+        q.Checksumming.store(checksumming);
         Y_ABORT_UNLESS(costModel);
         if (!q.CostModel || *q.CostModel != *costModel) {
             updated = true;
             q.CostModel = costModel;
         }
     } else {
         ConnectedQueuesMask[orderNumber] &= ~(1 << queueId);
-        q.ExtraBlockChecksSupport.reset();
-        q.Checksumming.reset();
+        q.ExtraBlockChecksSupport.store(false);
+        q.Checksumming.store(false);
         if (q.CostModel) {
             updated = true;
             q.CostModel = nullptr;

ydb/core/blobstorage/dsproxy/group_sessions.h

@@ -21,8 +21,20 @@ namespace NKikimr {
                 struct TQueue {
                     TActorId ActorId;
                     TIntrusivePtr<NBackpressure::TFlowRecord> FlowRecord;
-                    std::optional<bool> ExtraBlockChecksSupport;
-                    std::optional<bool> Checksumming;
+                    struct AtomicParameter : public std::atomic<bool> {
+                        AtomicParameter& operator=(const AtomicParameter& other) {
+                            store(other.load());
+                            return *this;
+                        }
+
+                        AtomicParameter(const AtomicParameter& other) {
+                            store(other.load());
+                        }
+
+                        AtomicParameter() {
+                            store(false);
+                        }
+                    } ExtraBlockChecksSupport, Checksumming;
                     std::shared_ptr<const TCostModel> CostModel = nullptr;
                     volatile bool IsConnected = false;
                 };
@@ -190,7 +202,7 @@ namespace NKikimr {
                 .Queues
                 .GetQueue(queueId)
                 .Checksumming
-                .value_or(false);
+                .load();
         }
 
         ui64 GetPredictedDelayNsByOrderNumber(ui32 orderNumber, NKikimrBlobStorage::EVDiskQueueId queueId) {