Some fixes in Bridge GRPC disconnect (#26493)

Author: mregrock

ydb/core/driver_lib/run/grpc_servers_manager.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <ydb/core/base/defs.h>
+#include <ydb/core/base/appdata.h>
+#include <ydb/library/actors/core/event_local.h>
+
+namespace NKikimr {
+
+struct TEvGRpcServersManager {
+    enum EEv {
+        EvDisconnectRequestStarted = EventSpaceBegin(TEvents::ES_PRIVATE) + 10000,
+        EvDisconnectRequestFinished,
+    };
+
+    struct TEvDisconnectRequestStarted : public TEventLocal<TEvDisconnectRequestStarted, EvDisconnectRequestStarted> {
+    };
+
+    struct TEvDisconnectRequestFinished : public TEventLocal<TEvDisconnectRequestFinished, EvDisconnectRequestFinished> {
+    };
+};
+
+inline TActorId MakeGRpcServersManagerId(ui32 nodeId) {
+    char x[12] = {'g','r','p','c','s','r','v','r','m','n','g','r'};
+    return TActorId(nodeId, TStringBuf(x, 12));
+}
+
+} // namespace NKikimr
+

ydb/core/driver_lib/run/run.cpp

@@ -1,5 +1,6 @@
 #include "auto_config_initializer.h"
 #include "run.h"
+#include "grpc_servers_manager.h"
 #include "service_initializer.h"
 #include "kikimr_services_initializers.h"
 
@@ -196,6 +197,8 @@ class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
     std::weak_ptr<TGRpcServersWrapper> GRpcServersWrapper;
     TIntrusivePtr<NMemory::IProcessMemoryInfoProvider> ProcessMemoryInfoProvider;
     bool Started = false;
+    bool StopScheduled = false;
+    bool WaitingForDisconnectRequest = false;
 
 public:
     enum {
@@ -227,17 +230,36 @@ class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
         if (const auto& bridgeInfo = ev->Get()->BridgeInfo) {
             if (NBridge::PileStateTraits(bridgeInfo->SelfNodePile->State).RequiresConfigQuorum) {
                 Start();
-            } else {
-                Stop();
+            } else if (!StopScheduled) {
+                StopScheduled = true;
+                CheckAndExecuteStop();
             }
         }
     }
 
+    void HandleDisconnectRequestStarted() {
+        WaitingForDisconnectRequest = true;
+    }
+
+    void HandleDisconnectRequestFinished() {
+        WaitingForDisconnectRequest = false;
+        CheckAndExecuteStop();
+    }
+
+    void CheckAndExecuteStop() {
+        if (StopScheduled && !WaitingForDisconnectRequest) {
+            StopScheduled = false;
+            Stop();
+        }
+    }
+
     void Start() {
         if (Started) {
             return;
         }
         Started = true;
+        StopScheduled = false;
+        WaitingForDisconnectRequest = false;
         auto wrapper = GRpcServersWrapper.lock();
         if (!wrapper) {
             return;
@@ -292,6 +314,8 @@ class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
     STRICT_STFUNC(StateFunc,
         hFunc(TEvNodeWardenStorageConfig, Handle)
         hFunc(TEvStop, HandleStop)
+        cFunc(TEvGRpcServersManager::EvDisconnectRequestStarted, HandleDisconnectRequestStarted)
+        cFunc(TEvGRpcServersManager::EvDisconnectRequestFinished, HandleDisconnectRequestFinished)
     )
 };
 
@@ -2020,6 +2044,7 @@ void TKikimrRunner::KikimrStart() {
     if (GRpcServersWrapper) {
         GRpcServersWrapper->Servers = GRpcServersWrapper->GrpcServersFactory();
         GRpcServersManager = ActorSystem->Register(new TGRpcServersManager(GRpcServersWrapper, ProcessMemoryInfoProvider));
+        ActorSystem->RegisterLocalService(NKikimr::MakeGRpcServersManagerId(ActorSystem->NodeId), GRpcServersManager);
     }
 
     if (SqsHttp) {

ydb/core/driver_lib/run/ya.make

@@ -14,6 +14,7 @@ SRCS(
     driver.h
     factories.h
     factories.cpp
+    grpc_servers_manager.h
     kikimr_services_initializers.cpp
     kikimr_services_initializers.h
     main.h

ydb/core/grpc_services/rpc_bridge.cpp

@@ -6,6 +6,7 @@
 #include <ydb/core/base/auth.h>
 #include <ydb/core/base/bridge.h>
 #include <ydb/core/base/blobstorage_common.h>
+#include <ydb/core/driver_lib/run/grpc_servers_manager.h>
 
 namespace NKikimr::NGRpcService {
 
@@ -59,6 +60,8 @@ class TUpdateClusterStateRequest : public TBridgeRequestGrpc<TUpdateClusterState
     using TBase = TBridgeRequestGrpc<TUpdateClusterStateRequest, TEvUpdateClusterStateRequest, Ydb::Bridge::UpdateClusterStateResult>;
     using TRpcBase = TRpcOperationRequestActor<TUpdateClusterStateRequest, TEvUpdateClusterStateRequest>;
 
+    bool IsDisconnectingRequest = false;
+
 public:
     using TBase::TBase;
 
@@ -248,6 +251,19 @@ class TUpdateClusterStateRequest : public TBridgeRequestGrpc<TUpdateClusterState
         finalPromoted->CopyToProto(&newClusterState, &NKikimrBridge::TClusterState::SetPromotedPile);
         newClusterState.SetGeneration(currentClusterState.GetGeneration() + 1);
 
+        // check if current node is transitioning to DISCONNECTED state
+        // if so, notify GRpcServersManager to wait for this request before shutting down
+        const auto selfNodePileId = bridgeInfo->SelfNodePile->BridgePileId;
+        if (selfNodePileId.GetPileIndex() < newClusterState.PerPileStateSize()) {
+            const auto newSelfState = newClusterState.GetPerPileState(selfNodePileId.GetPileIndex());
+            if (newSelfState == NKikimrBridge::TClusterState::DISCONNECTED ||
+                newSelfState == NKikimrBridge::TClusterState::SUSPENDED) {
+                IsDisconnectingRequest = true;
+                self->ActorContext().Send(NKikimr::MakeGRpcServersManagerId(self->SelfId().NodeId()),
+                    new NKikimr::TEvGRpcServersManager::TEvDisconnectRequestStarted());
+            }
+        }
+
         auto request = std::make_unique<NStorage::TEvNodeConfigInvokeOnRoot>();
         auto *cmd = request->Record.MutableSwitchBridgeClusterState();
         cmd->MutableNewClusterState()->CopyFrom(newClusterState);
@@ -280,13 +296,23 @@ class TUpdateClusterStateRequest : public TBridgeRequestGrpc<TUpdateClusterState
     void Handle(NStorage::TEvNodeConfigInvokeOnRootResult::TPtr& ev) {
         const auto& response = ev->Get()->Record;
         auto* self = Self();
-        if (response.GetStatus() == NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult::OK) {
-            Ydb::Bridge::UpdateClusterStateResult result;
-            self->ReplyWithResult(Ydb::StatusIds::SUCCESS, result, self->ActorContext());
+        const bool success = response.GetStatus() == NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult::OK;
+        auto notifyDisconnectFinished = [&]() {
+            if (IsDisconnectingRequest) {
+                self->ActorContext().Send(NKikimr::MakeGRpcServersManagerId(self->SelfId().NodeId()),
+                    new NKikimr::TEvGRpcServersManager::TEvDisconnectRequestFinished());
+            }
+        };
+
+        if (!success) {
+            notifyDisconnectFinished();
+            self->Reply(Ydb::StatusIds::INTERNAL_ERROR, response.GetErrorReason(),
+                NKikimrIssues::TIssuesIds::DEFAULT_ERROR, self->ActorContext());
             return;
         }
-        self->Reply(Ydb::StatusIds::INTERNAL_ERROR, response.GetErrorReason(),
-            NKikimrIssues::TIssuesIds::DEFAULT_ERROR, self->ActorContext());
+        Ydb::Bridge::UpdateClusterStateResult result;
+        self->ReplyWithResult(Ydb::StatusIds::SUCCESS, result, self->ActorContext());
+        notifyDisconnectFinished();
     }
 };
 

ydb/tests/functional/bridge/test_bridge.py

@@ -30,12 +30,6 @@ def test_failover(self):
         self.update_cluster_state(self.bridge_client, updates)
         self.wait_for_cluster_state(self.secondary_bridge_client, {"r1": PileState.DISCONNECTED, "r2": PileState.PRIMARY})
 
-        updates = [
-            PileState(pile_name="r1", state=PileState.NOT_SYNCHRONIZED),
-        ]
-        self.update_cluster_state(self.secondary_bridge_client, updates)
-        self.wait_for_cluster_state(self.secondary_bridge_client, {"r1": PileState.SYNCHRONIZED, "r2": PileState.PRIMARY}, timeout_seconds=50)
-
     def test_takedown(self):
         initial_result = self.get_cluster_state(self.bridge_client)
         self.check_states(initial_result, {"r1": PileState.PRIMARY, "r2": PileState.SYNCHRONIZED})
@@ -44,11 +38,6 @@ def test_takedown(self):
         ]
         self.update_cluster_state(self.bridge_client, updates)
         self.wait_for_cluster_state(self.bridge_client, {"r1": PileState.PRIMARY, "r2": PileState.DISCONNECTED})
-        updates = [
-            PileState(pile_name="r2", state=PileState.NOT_SYNCHRONIZED),
-        ]
-        self.update_cluster_state(self.bridge_client, updates)
-        self.wait_for_cluster_state(self.bridge_client, {"r1": PileState.PRIMARY, "r2": PileState.SYNCHRONIZED}, timeout_seconds=50)
 
     def test_rejoin(self):
         initial_result = self.get_cluster_state(self.bridge_client)