Fix DDisk shutdown lifecycle (#46775)

5 files changed, 214 insertions, 2 deletions

diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
index c2ff4253a09..b4571a4f566 100644
--- a/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
+++ b/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
@@ -413,6 +413,9 @@ namespace {
         }
 #endif
         CountersBase->RemoveSubgroupChain(CountersChain);
+        if (!IsPersistentBufferActor) {
+            Send(MakeBlobStorageNodeWardenID(SelfId().NodeId()), new TEvents::TEvGone());
+        }
         TActorBootstrapped::PassAway();
     }
 
diff --git a/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp b/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp
index 4a15c5539f6..c9c5a9b69f3 100644
--- a/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp
+++ b/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp
@@ -306,6 +306,54 @@ std::unique_ptr<TEventHandle<NDDisk::TEvWriteResult>> DoWrite(TTestContext& ctx,
 } // anonymous namespace
 
 Y_UNIT_TEST_SUITE(TDDiskActorTest) {
+    Y_UNIT_TEST(PoisonNotifiesNodeWardenImmediately) {
+        TTestContext ctx;
+        ctx.Runtime.RegisterService(MakeBlobStorageNodeWardenID(NodeId), ctx.Edge);
+
+        const TDiskHandle disk = ctx.CreateDDisk(43, 1);
+        const TActorId ddiskActorId =
+            ctx.Runtime.GetNode(NodeId)->ActorSystem->LookupLocalService(disk.ServiceId);
+        const TActorId persistentBufferActorId =
+            ctx.Runtime.GetNode(NodeId)->ActorSystem->LookupLocalService(disk.PBServiceId);
+        UNIT_ASSERT(ddiskActorId);
+        UNIT_ASSERT(persistentBufferActorId);
+
+        std::unique_ptr<IEventHandle> blockedPersistentBufferPoison;
+        ctx.Runtime.FilterFunction = [&](ui32 /*nodeId*/, std::unique_ptr<IEventHandle>& ev) {
+            if (!blockedPersistentBufferPoison &&
+                    ev->GetTypeRewrite() == TEvents::TSystem::Poison &&
+                    ev->Recipient == persistentBufferActorId) {
+                blockedPersistentBufferPoison = std::move(ev);
+                return false;
+            }
+            return true;
+        };
+
+        SendToDDisk(ctx, disk.ServiceId, new TEvents::TEvPoison());
+        ui32 eventsProcessed = 0;
+        ctx.Runtime.Sim([&] {
+            return !blockedPersistentBufferPoison && ++eventsProcessed <= 200;
+        });
+        UNIT_ASSERT_C(blockedPersistentBufferPoison, "DDisk must poison its persistent buffer actor");
+        UNIT_ASSERT(!ctx.Runtime.WrapInActorContext(ddiskActorId, [](IActor*) {}));
+        UNIT_ASSERT(ctx.Runtime.WrapInActorContext(persistentBufferActorId, [](IActor*) {}));
+
+        ctx.Runtime.FilterFunction = {};
+        const auto gone = WaitFromDDisk<TEvents::TEvGone>(ctx);
+
+        ctx.Runtime.Send(std::move(blockedPersistentBufferPoison), NodeId);
+
+        UNIT_ASSERT_VALUES_EQUAL(gone->Sender, ddiskActorId);
+        UNIT_ASSERT(!ctx.Runtime.WrapInActorContext(ddiskActorId, [](IActor*) {}));
+        ui32 persistentBufferEventsProcessed = 0;
+        ctx.Runtime.Sim([&] {
+            return ctx.Runtime.WrapInActorContext(persistentBufferActorId, [](IActor*) {})
+                && ++persistentBufferEventsProcessed <= 200;
+        });
+        UNIT_ASSERT_C(!ctx.Runtime.WrapInActorContext(persistentBufferActorId, [](IActor*) {}),
+            "Persistent buffer must stop after receiving poison");
+    }
+
     Y_UNIT_TEST(SessionValidation) {
         TTestContext ctx;
         const TDiskHandle disk = ctx.CreateDDisk(1, 1);
diff --git a/ydb/core/blobstorage/nodewarden/blobstorage_node_warden_ut.cpp b/ydb/core/blobstorage/nodewarden/blobstorage_node_warden_ut.cpp
index a93e6cce7ab..bcdda630bf5 100644
--- a/ydb/core/blobstorage/nodewarden/blobstorage_node_warden_ut.cpp
+++ b/ydb/core/blobstorage/nodewarden/blobstorage_node_warden_ut.cpp
@@ -1408,6 +1408,163 @@ Y_UNIT_TEST_SUITE(TBlobStorageWardenTest) {
         CheckVDiskStateUpdate(runtime, fakeWhiteboard, groupId, 2, 2u);
     }
 
+    class TSilentPDiskActor : public TActorBootstrapped<TSilentPDiskActor> {
+    public:
+        void Bootstrap() {
+            Become(&TThis::StateFunc);
+        }
+
+        STFUNC(StateFunc) {
+            if (ev->GetTypeRewrite() == TEvents::TSystem::Poison) {
+                PassAway();
+            }
+        }
+    };
+
+    class TSilentPDiskServiceFactory : public IPDiskServiceFactory {
+    public:
+        void Create(const TActorContext& ctx, ui32 pdiskId, const TIntrusivePtr<TPDiskConfig>&,
+                const NPDisk::TMainKey&, ui32 poolId, ui32 nodeId) override {
+            const TActorId actorId = ctx.Register(new TSilentPDiskActor, TMailboxType::HTSwap, poolId);
+            ctx.ActorSystem()->RegisterLocalService(MakeBlobStoragePDiskID(nodeId, pdiskId), actorId);
+        }
+    };
+
+    struct TDDiskLifecycleTestSetup {
+        static constexpr ui32 NodeId = 1;
+        static constexpr ui32 PDiskId = 1;
+        static constexpr ui32 VDiskSlotId = 1;
+
+        TTestActorSystem Runtime;
+        const ui32 GroupId;
+        const TVDiskID VDiskId;
+        const TActorId DDiskServiceId;
+        TActorId NodeWardenId;
+
+        TDDiskLifecycleTestSetup()
+            : Runtime(1, NLog::PRI_ERROR, MakeIntrusive<TDomainsInfo>())
+            , GroupId(TGroupID(EGroupConfigurationType::Dynamic, 1, 1).GetRaw())
+            , VDiskId(GroupId, 1, 0, 0, 0)
+            , DDiskServiceId(MakeBlobStorageDDiskId(NodeId, PDiskId, VDiskSlotId))
+        {
+            Runtime.Start();
+
+            auto& appData = *Runtime.GetNode(NodeId)->AppData;
+            appData.DomainsInfo->AddDomain(TDomainsInfo::TDomain::ConstructEmptyDomain("dom", 1).Release());
+            appData.DynamicNameserviceConfig = new TDynamicNameserviceConfig();
+            appData.DynamicNameserviceConfig->MaxStaticNodeId = NodeId;
+
+            TIntrusivePtr<TNodeWardenConfig> nodeWardenConfig(
+                new TNodeWardenConfig(new TSilentPDiskServiceFactory));
+            nodeWardenConfig->DDiskConfig.emplace();
+            nodeWardenConfig->DDiskConfig->SetForcePDiskFallback(true);
+
+            auto* serviceSet = nodeWardenConfig->BlobStorageConfig.MutableServiceSet();
+            auto* pdisk = serviceSet->AddPDisks();
+            pdisk->SetNodeID(NodeId);
+            pdisk->SetPDiskID(PDiskId);
+            pdisk->SetPath("silent-pdisk");
+            pdisk->SetPDiskGuid(12345);
+            pdisk->SetPDiskCategory(TPDiskCategory(NPDisk::DEVICE_TYPE_NVME, 0).GetRaw());
+
+            auto* group = serviceSet->AddGroups();
+            FillGroup(group, VDiskId.GroupGeneration);
+
+            FillDDisk(serviceSet->AddVDisks());
+
+            NodeWardenId = Runtime.Register(CreateBSNodeWarden(nodeWardenConfig.Release()), NodeId);
+            Runtime.RegisterService(MakeBlobStorageNodeWardenID(NodeId), NodeWardenId);
+            UNIT_ASSERT(Runtime.WrapInActorContext(NodeWardenId, [](IActor* actor) {
+                dynamic_cast<NStorage::TNodeWarden*>(actor)->Bootstrap();
+            }));
+            UNIT_ASSERT(LookupDDiskActor());
+        }
+
+        ~TDDiskLifecycleTestSetup() {
+            Runtime.Stop();
+        }
+
+        void FillGroup(NKikimrBlobStorage::TGroupInfo* group, ui32 generation) const {
+            group->SetGroupID(GroupId);
+            group->SetGroupGeneration(generation);
+            group->SetErasureSpecies(TBlobStorageGroupType::ErasureNone);
+            group->SetStoragePoolName("ddisk-pool");
+            group->SetDDisk(true);
+            auto* location = group->AddRings()->AddFailDomains()->AddVDiskLocations();
+            FillLocation(location);
+        }
+
+        void FillLocation(NKikimrBlobStorage::TVDiskLocation* location) const {
+            location->SetNodeID(NodeId);
+            location->SetPDiskID(PDiskId);
+            location->SetVDiskSlotID(VDiskSlotId);
+            location->SetPDiskGuid(12345);
+        }
+
+        void FillDDisk(NKikimrBlobStorage::TNodeWardenServiceSet::TVDisk* vdisk) const {
+            VDiskIDFromVDiskID(VDiskId, vdisk->MutableVDiskID());
+            FillLocation(vdisk->MutableVDiskLocation());
+            vdisk->SetStoragePoolName("ddisk-pool");
+        }
+
+        TActorId LookupDDiskActor() {
+            return Runtime.GetNode(NodeId)->ActorSystem->LookupLocalService(DDiskServiceId);
+        }
+
+        bool IsActorAlive(const TActorId& actorId) {
+            return Runtime.WrapInActorContext(actorId, [](IActor*) {});
+        }
+
+        template<typename TPredicate>
+        void DispatchUntil(TPredicate&& predicate, TStringBuf description) {
+            ui32 eventsProcessed = 0;
+            Runtime.Sim([&] {
+                return !predicate() && ++eventsProcessed <= 200;
+            });
+            UNIT_ASSERT_C(predicate(), description);
+        }
+
+        void DeleteDDisk() {
+            UNIT_ASSERT(Runtime.WrapInActorContext(NodeWardenId, [&](IActor* actor) {
+                NKikimrBlobStorage::TNodeWardenServiceSet serviceSet;
+                auto* vdisk = serviceSet.AddVDisks();
+                FillDDisk(vdisk);
+                vdisk->SetEntityStatus(NKikimrBlobStorage::DESTROY);
+                dynamic_cast<NStorage::TNodeWarden*>(actor)->ApplyServiceSet(
+                    serviceSet, true, false, false, "test");
+            }));
+        }
+
+        void RestartDDisk() {
+            Runtime.Send(new IEventHandle(NodeWardenId, {}, new TEvBlobStorage::TEvAskRestartVDisk(PDiskId, VDiskId)),
+                NodeId);
+        }
+
+    };
+
+    Y_UNIT_TEST(TestDDiskDeleteStopsRunningActor) {
+        TDDiskLifecycleTestSetup setup;
+        const TActorId actorId = setup.LookupDDiskActor();
+
+        setup.DeleteDDisk();
+        setup.DispatchUntil([&] { return !setup.IsActorAlive(actorId); },
+            "DDisk actor must stop after its VDisk is deleted");
+    }
+
+    Y_UNIT_TEST(TestDDiskGoneAllowsRestart) {
+        TDDiskLifecycleTestSetup setup;
+        const TActorId previousActorId = setup.LookupDDiskActor();
+
+        setup.RestartDDisk();
+        setup.DispatchUntil([&] {
+            const TActorId currentActorId = setup.LookupDDiskActor();
+            return currentActorId && currentActorId != previousActorId;
+        }, "NodeWarden must restart DDisk after receiving TEvGone");
+
+        UNIT_ASSERT(!setup.IsActorAlive(previousActorId));
+        UNIT_ASSERT(setup.IsActorAlive(setup.LookupDDiskActor()));
+    }
+
     struct TStaticGroupProxyTestSetup {
         TTestActorSystem Runtime;
         ui32 NodeId = 1;
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_impl.h b/ydb/core/blobstorage/nodewarden/node_warden_impl.h
index 794d63cb68c..1416551804a 100644
--- a/ydb/core/blobstorage/nodewarden/node_warden_impl.h
+++ b/ydb/core/blobstorage/nodewarden/node_warden_impl.h
@@ -459,6 +459,8 @@ namespace NKikimr::NStorage {
             // Runtime configuration of VDisk.
             struct TRuntimeData {
                 TIntrusivePtr<TBlobStorageGroupInfo> GroupInfo;
+                TActorId ActorId;
+                TActorId ServiceId;
                 ui32 OrderNumber;
                 bool DonorMode;
                 bool ReadOnly;
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp b/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
index 556df00aa88..fb7090e9d85 100644
--- a/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
+++ b/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
@@ -39,7 +39,7 @@ namespace NKikimr::NStorage {
         if (vdisk.RuntimeData) {
             vdiskRunning = true;
             vdisk.TIntrusiveListItem<TVDiskRecord, TGroupRelationTag>::Unlink();
-            TActivationContext::Send(new IEventHandle(TEvents::TSystem::Poison, 0, vdisk.GetVDiskServiceId(), {}, nullptr, 0));
+            TActivationContext::Send(new IEventHandle(TEvents::TSystem::Poison, 0, vdisk.RuntimeData->ActorId, {}, nullptr, 0));
             vdisk.RuntimeData.reset();
         }
 
@@ -430,6 +430,8 @@ namespace NKikimr::NStorage {
 
         vdisk.RuntimeData.emplace(TVDiskRecord::TRuntimeData{
             .GroupInfo = groupInfo,
+            .ActorId = actorId,
+            .ServiceId = vdiskServiceId,
             .OrderNumber = groupInfo->GetOrderNumber(TVDiskIdShort(vdiskId)),
             .DonorMode = donorMode,
             .ReadOnly = readOnly,
@@ -601,7 +603,7 @@ namespace NKikimr::NStorage {
         Y_ABORT_UNLESS(newInfo->GroupID == currentInfo->GroupID);
 
         const ui32 orderNumber = vdisk.RuntimeData->OrderNumber;
-        const TActorId vdiskServiceId = vdisk.GetVDiskServiceId();
+        const TActorId vdiskServiceId = vdisk.RuntimeData->ServiceId;
 
         if (newInfo->GetActorId(orderNumber) != vdiskServiceId) {
             // this disk is in donor mode, we don't care about generation change; donor modes are operated by BSC solely