Fix distconf IC session subscription state machine (#7143)

Author: alexvru

.github/config/muted_ya.txt

@@ -1,6 +1,5 @@
 ydb/core/blobstorage/dsproxy/ut TBlobStorageProxySequenceTest.TestBlock42PutWithChangingSlowDisk
 ydb/core/blobstorage/dsproxy/ut_fat TBlobStorageProxyTest.TestBatchedPutRequestDoesNotContainAHugeBlob
-ydb/core/blobstorage/ut_blobstorage BlobPatching.PatchBlock42
 ydb/core/client/ut TClientTest.PromoteFollower
 ydb/core/client/ut TClientTest.ReadFromFollower
 ydb/core/client/ut TFlatTest.AutoSplitMergeQueue

ydb/core/blobstorage/nodewarden/distconf.cpp

@@ -145,30 +145,44 @@ namespace NKikimr::NStorage {
             }
         }
 
-        for (const auto& [nodeId, sessionId] : SubscribedSessions) {
+        for (const auto& [nodeId, subs] : SubscribedSessions) {
             bool okay = false;
             if (Binding && Binding->NodeId == nodeId) {
-                Y_ABORT_UNLESS(sessionId == Binding->SessionId);
+                Y_VERIFY_S(subs.SessionId == Binding->SessionId || !Binding->SessionId,
+                    "Binding# " << Binding->ToString() << " Subscription# " << subs.ToString());
                 okay = true;
             }
             if (const auto it = DirectBoundNodes.find(nodeId); it != DirectBoundNodes.end()) {
-                Y_ABORT_UNLESS(!sessionId || sessionId == it->second.SessionId);
+                Y_VERIFY_S(!subs.SessionId || subs.SessionId == it->second.SessionId, "sessionId# " << subs.SessionId
+                    << " node.SessionId# " << it->second.SessionId);
                 okay = true;
             }
-            if (!sessionId) {
+            if (!subs.SessionId) {
                 okay = true; // may be just obsolete subscription request
             }
             if (ConnectedDynamicNodes.contains(nodeId)) {
                 okay = true;
             }
             Y_ABORT_UNLESS(okay);
+            if (subs.SubscriptionCookie) {
+                const auto it = SubscriptionCookieMap.find(subs.SubscriptionCookie);
+                Y_ABORT_UNLESS(it != SubscriptionCookieMap.end());
+                Y_ABORT_UNLESS(it->second == nodeId);
+            }
+        }
+        for (const auto& [cookie, nodeId] : SubscriptionCookieMap) {
+            const auto it = SubscribedSessions.find(nodeId);
+            Y_ABORT_UNLESS(it != SubscribedSessions.end());
+            const TSessionSubscription& subs = it->second;
+            Y_VERIFY_S(subs.SubscriptionCookie == cookie, "SubscriptionCookie# " << subs.SubscriptionCookie
+                << " cookie# " << cookie);
         }
 
         if (Binding) {
             Y_ABORT_UNLESS(SubscribedSessions.contains(Binding->NodeId));
         }
         for (const auto& [nodeId, info] : DirectBoundNodes) {
-            Y_ABORT_UNLESS(SubscribedSessions.contains(nodeId));
+            Y_VERIFY_S(SubscribedSessions.contains(nodeId), "NodeId# " << nodeId);
         }
 
         Y_ABORT_UNLESS(!StorageConfig || CheckFingerprint(*StorageConfig));
@@ -229,6 +243,13 @@ namespace NKikimr::NStorage {
     }
 
     STFUNC(TDistributedConfigKeeper::StateFunc) {
+        STLOG(PRI_DEBUG, BS_NODE, NWDC15, "StateFunc", (Type, ev->GetTypeRewrite()), (Sender, ev->Sender),
+            (SessionId, ev->InterconnectSession), (Cookie, ev->Cookie));
+        const ui32 senderNodeId = ev->Sender.NodeId();
+        if (ev->InterconnectSession && SubscribedSessions.contains(senderNodeId)) {
+            // keep session actors intact
+            SubscribeToPeerNode(senderNodeId, ev->InterconnectSession);
+        }
         STRICT_STFUNC_BODY(
             hFunc(TEvNodeConfigPush, Handle);
             hFunc(TEvNodeConfigReversePush, Handle);
@@ -239,7 +260,6 @@ namespace NKikimr::NStorage {
             hFunc(TEvInterconnect::TEvNodesInfo, Handle);
             hFunc(TEvInterconnect::TEvNodeConnected, Handle);
             hFunc(TEvInterconnect::TEvNodeDisconnected, Handle);
-            hFunc(TEvents::TEvUndelivered, Handle);
             cFunc(TEvPrivate::EvErrorTimeout, HandleErrorTimeout);
             hFunc(TEvPrivate::TEvStorageConfigLoaded, Handle);
             hFunc(TEvPrivate::TEvStorageConfigStored, Handle);
@@ -252,6 +272,12 @@ namespace NKikimr::NStorage {
             cFunc(TEvents::TSystem::Wakeup, HandleWakeup);
             cFunc(TEvents::TSystem::Poison, PassAway);
         )
+        for (ui32 nodeId : std::exchange(UnsubscribeQueue, {})) {
+            UnsubscribeInterconnect(nodeId);
+        }
+        if (IsSelfStatic && StorageConfig && NodeListObtained) {
+            IssueNextBindRequest();
+        }
         ConsistencyCheck();
     }
 

ydb/core/blobstorage/nodewarden/distconf.h

@@ -240,7 +240,20 @@ namespace NKikimr::NStorage {
         TString ErrorReason;
 
         // subscribed IC sessions
-        THashMap<ui32, TActorId> SubscribedSessions;
+        struct TSessionSubscription {
+            TActorId SessionId;
+            ui64 SubscriptionCookie = 0; // when nonzero, we didn't have TEvNodeConnected yet
+
+            TSessionSubscription(TActorId sessionId) : SessionId(sessionId) {}
+
+            TString ToString() const {
+                return TStringBuilder() << "{SessionId# " << SessionId << " SubscriptionCookie# " << SubscriptionCookie << "}";
+            }
+        };
+        THashMap<ui32, TSessionSubscription> SubscribedSessions;
+        ui64 NextSubscribeCookie = 1;
+        THashMap<ui64, ui32> SubscriptionCookieMap;
+        THashSet<ui32> UnsubscribeQueue;
 
         // child actors
         THashSet<TActorId> ChildActors;
@@ -287,8 +300,9 @@ namespace NKikimr::NStorage {
         void BindToSession(TActorId sessionId);
         void Handle(TEvInterconnect::TEvNodeConnected::TPtr ev);
         void Handle(TEvInterconnect::TEvNodeDisconnected::TPtr ev);
-        void Handle(TEvents::TEvUndelivered::TPtr ev);
+        void HandleDisconnect(ui32 nodeId, TActorId sessionId);
         void UnsubscribeInterconnect(ui32 nodeId);
+        TActorId SubscribeToPeerNode(ui32 nodeId, TActorId sessionId);
         void AbortBinding(const char *reason, bool sendUnbindMessage = true);
         void HandleWakeup();
         void Handle(TEvNodeConfigReversePush::TPtr ev);