Add ExplicitMultiPut event to avoid data races in DSProxy UT (#25761)

Author: serbel324

ydb/core/base/blobstorage.h

@@ -496,6 +496,8 @@ struct TEvBlobStorage {
         EvGetBlock,
         EvCheckIntegrity,
 
+        EvExplicitMultiPut, // for debugging purposes
+
         //
         EvPutResult = EvPut + 512,                              /// 268 632 576
         EvGetResult,

ydb/core/blobstorage/dsproxy/dsproxy.h

@@ -195,6 +195,7 @@ class TBlobStorageGroupRequestActor : public TActor<TBlobStorageGroupRequestActo
         TMaybe<TGroupStat::EKind> LatencyQueueKind = {};
 
         std::optional<ui32> ForceGroupGeneration; // work only with this specific group generation and nothing else
+        bool DoSendDeathNote = true; // unschedules DSProxy timeout on termination, be careful with disabling
     };
 
     struct TTypeSpecificParameters {
@@ -222,6 +223,7 @@ class TBlobStorageGroupRequestActor : public TActor<TBlobStorageGroupRequestActo
         , RacingDomains(&Info->GetTopology())
         , ExecutionRelay(std::move(params.Common.ExecutionRelay))
         , ForceGroupGeneration(params.Common.ForceGroupGeneration)
+        , DoSendDeathNote(params.Common.DoSendDeathNote)
     {
         if (ParentSpan) {
             const NWilson::TTraceId& parentTraceId = ParentSpan.GetTraceId();
@@ -327,6 +329,7 @@ class TBlobStorageGroupRequestActor : public TActor<TBlobStorageGroupRequestActo
     bool FirstResponse = true;
     std::optional<ui32> ForceGroupGeneration;
     ui32 RacingGeneration = 0;
+    bool DoSendDeathNote;
 };
 
 void Encrypt(char *destination, const char *source, size_t shift, size_t sizeBytes, const TLogoBlobID &id,

ydb/core/blobstorage/dsproxy/dsproxy_impl.h

@@ -2,6 +2,7 @@
 
 #include "defs.h"
 #include "dsproxy.h"
+#include "dsproxy_test_helpers.h"
 
 #include <ydb/core/blobstorage/base/utility.h>
 
@@ -294,6 +295,7 @@ class TBlobStorageGroupProxy : public TActorBootstrapped<TBlobStorageGroupProxy>
     void Handle(TEvBlobStorage::TEvBunchOfEvents::TPtr ev);
     void Handle(TEvDeathNote::TPtr ev);
     void Handle(TEvGetQueuesInfo::TPtr ev);
+    void Handle(TEvExplicitMultiPut::TPtr ev);
 
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     // Error state
@@ -388,6 +390,7 @@ class TBlobStorageGroupProxy : public TActorBootstrapped<TBlobStorageGroupProxy>
         fFunc(Ev5min, Handle5min);
         cFunc(EvCheckDeadlines, CheckDeadlines);
         hFunc(TEvGetQueuesInfo, Handle);
+        hFunc(TEvExplicitMultiPut, Handle);
     )
 
 #define HANDLE_EVENTS(HANDLER) \

ydb/core/blobstorage/dsproxy/dsproxy_request.cpp

@@ -918,7 +918,9 @@ namespace NKikimr {
         // ensure that we are dying for the first time
         Y_ABORT_UNLESS(!std::exchange(Dead, true));
         GetActiveCounter()->Dec();
-        SendToProxy(std::make_unique<TEvDeathNote>(Responsiveness));
+        if (DoSendDeathNote) {
+            SendToProxy(std::make_unique<TEvDeathNote>(Responsiveness));
+        }
         TActor::PassAway();
     }
 
@@ -1137,4 +1139,9 @@ namespace NKikimr {
         TActivationContext::Send(ev->Sender, std::move(res));
     }
 
+    void TBlobStorageGroupProxy::Handle(TEvExplicitMultiPut::TPtr ev) {
+        IActor *reqActor = CreateBlobStorageGroupPutRequest(std::move(ev->Get()->Parameters));
+        TActivationContext::Register(reqActor);
+    }
+
 } // NKikimr

ydb/core/blobstorage/dsproxy/dsproxy_test_helpers.h

@@ -0,0 +1,18 @@
+#pragma once
+
+#include "defs.h"
+#include "dsproxy.h"
+
+namespace NKikimr {
+
+struct TEvExplicitMultiPut : public TEventLocal<TEvExplicitMultiPut, TEvBlobStorage::EvExplicitMultiPut> {
+public:
+    TEvExplicitMultiPut(TBlobStorageGroupMultiPutParameters params)
+        : Parameters(params)
+    {}
+
+public:
+    TBlobStorageGroupMultiPutParameters Parameters;
+};
+
+} // namespace NKikimr

ydb/core/blobstorage/dsproxy/ut_fat/dsproxy_ut.cpp

@@ -3,6 +3,7 @@
 #include <ydb/core/blobstorage/base/blobstorage_events.h>
 #include <ydb/core/blobstorage/dsproxy/dsproxy.h>
 #include <ydb/core/blobstorage/dsproxy/dsproxy_nodemon.h>
+#include <ydb/core/blobstorage/dsproxy/dsproxy_test_helpers.h>
 #include <ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.h>
 #include <ydb/core/blobstorage/pdisk/blobstorage_pdisk.h>
 #include <ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h>
@@ -3433,32 +3434,29 @@ class TTestBlobStorageProxyBatchedPutRequestDoesNotContainAHugeBlob : public TTe
 
         switch (TestStep) {
             case 0: {
-                TBatchedVec<TEvBlobStorage::TEvPut::TPtr> batched(2);
-                batched[0] = GetPut(blobIds[0], Data1);
-                batched[1] = GetPut(blobIds[1], Data2);
+                Batched[0] = GetPut(blobIds[0], Data1);
+                Batched[1] = GetPut(blobIds[1], Data2);
 
                 TMaybe<TGroupStat::EKind> kind = PutHandleClassToGroupStatKind(HandleClass);
-                IActor *reqActor = CreateBlobStorageGroupPutRequest(
-                        TBlobStorageGroupMultiPutParameters{
-                            .Common = {
-                                .GroupInfo = BsInfo,
-                                .GroupQueues = GroupQueues,
-                                .Mon = Mon,
-                                .Now = TMonotonic::Now(),
-                                .StoragePoolCounters = StoragePoolCounters,
-                                .RestartCounter = TBlobStorageGroupMultiPutParameters::CalculateRestartCounter(batched),
-                                .LatencyQueueKind = kind,
-                            },
-                            .Events = batched,
-                            .TimeStatsEnabled = false,
-                            .Stats = PerDiskStatsPtr,
-                            .HandleClass = HandleClass,
-                            .Tactic = Tactic,
-                            .EnableRequestMod3x3ForMinLatency = false,
-                            .AccelerationParams = TAccelerationParams{},
-                        });
-
-                ctx.Register(reqActor);
+                ctx.Send(Proxy, new TEvExplicitMultiPut(TBlobStorageGroupMultiPutParameters{
+                        .Common = {
+                            .GroupInfo = BsInfo,
+                            .GroupQueues = GroupQueues,
+                            .Mon = Mon,
+                            .Now = TMonotonic::Now(),
+                            .StoragePoolCounters = StoragePoolCounters,
+                            .RestartCounter = TBlobStorageGroupMultiPutParameters::CalculateRestartCounter(Batched),
+                            .LatencyQueueKind = kind,
+                            .DoSendDeathNote = false,
+                        },
+                        .Events = Batched,
+                        .TimeStatsEnabled = false,
+                        .Stats = PerDiskStatsPtr,
+                        .HandleClass = HandleClass,
+                        .Tactic = Tactic,
+                        .EnableRequestMod3x3ForMinLatency = false,
+                        .AccelerationParams = TAccelerationParams{},
+                }));
                 break;
             }
             case 10:
@@ -3486,10 +3484,12 @@ class TTestBlobStorageProxyBatchedPutRequestDoesNotContainAHugeBlob : public TTe
     NKikimrBlobStorage::EPutHandleClass HandleClass = NKikimrBlobStorage::TabletLog;
     TString Data1;
     TString Data2;
+    TBatchedVec<TEvBlobStorage::TEvPut::TPtr> Batched;
 public:
     TTestBlobStorageProxyBatchedPutRequestDoesNotContainAHugeBlob(const TActorId &proxy, const TIntrusivePtr<TBlobStorageGroupInfo> &bsInfo,
             const TIntrusivePtr<TTestEnvironment> &env, const TIntrusivePtr<ITestParametrs> &parametrs)
         : TTestBlobStorageProxyForRequest(proxy, bsInfo, env, parametrs)
+        , Batched(2)
     {
         Data1.resize(MaxBatchedPutSize - 1, 'a');
         Data2.resize(1, 'a');