Bridge Mode in SB: gracefully continue sync request execution in case of concurrent reconfiguration (#20712)

Author: jepett0

ydb/core/tx/scheme_board/subscriber.cpp

@@ -1040,6 +1040,13 @@ class TSubscriber: public TMonitorableActor<TDerived> {
                 this->Send(proxy, new TEvents::TEvPoisonPill());
             }
         }
+
+        if (CurrentSyncRequest) {
+            SBS_LOG_I("Delay current sync request: " << CurrentSyncRequest);
+            DelayedSyncRequest = Max(DelayedSyncRequest, CurrentSyncRequest);
+            CurrentSyncRequest = 0;
+        }
+
         ProxyToGroupMap.clear();
         ProxyGroups.clear();
         States.clear();

ydb/core/tx/scheme_board/subscriber_ut.cpp

@@ -850,6 +850,97 @@ Y_UNIT_TEST_SUITE(TSubscriberSyncQuorumTest) {
             UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Get()->Partial, true, syncResponse->ToString());
         }
     }
+
+    Y_UNIT_TEST(ReconfigurationWithDelayedSyncRequest) {
+        TTestBasicRuntime runtime;
+        SetupMinimalRuntime(runtime);
+
+        runtime.SetLogPriority(NKikimrServices::SCHEME_BOARD_SUBSCRIBER, NLog::PRI_DEBUG);
+
+        const auto stateStorageInfo = GetStateStorageInfo(runtime);
+        const auto participatingReplicas = CountParticipatingReplicas(*stateStorageInfo);
+
+        constexpr int DomainId = 1;
+        constexpr const char* Path = "TestPath";
+        const TActorId edge = runtime.AllocateEdgeActor();
+
+        const TActorId subscriber = runtime.Register(CreateSchemeBoardSubscriber(edge, Path, DomainId));
+        TBlockEvents<NInternalEvents::TEvNotify> notificationBlocker(runtime, [&](const NInternalEvents::TEvNotify::TPtr& ev) {
+            return ev->Recipient == subscriber;
+        });
+        runtime.WaitFor("initial path lookups", [&]() {
+            return notificationBlocker.size() == participatingReplicas;
+        }, TDuration::Seconds(10));
+
+        // Send sync request: subscriber will queue it in DelayedSyncRequest since it cannot process syncs before finishing its initialization.
+        constexpr ui64 cookie = 12345;
+        runtime.Send(new IEventHandle(subscriber, edge, new NInternalEvents::TEvSyncRequest(), 0, cookie));
+
+        auto replicas = ResolveReplicas(runtime, Path);
+        runtime.Send(subscriber, edge, replicas->Release().Release());
+
+        // Now allow all notifications through so that initialization completes.
+        notificationBlocker.Stop().Unblock();
+
+        auto syncResponse = runtime.GrabEdgeEvent<NInternalEvents::TEvSyncResponse>(edge, TDuration::Seconds(10));
+        UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Get()->Path, Path, syncResponse->ToString());
+        UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Cookie, cookie, syncResponse->ToString());
+        UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Get()->Partial, false, syncResponse->ToString());
+
+        // No additional sync responses.
+        UNIT_CHECK_GENERATED_EXCEPTION(
+            runtime.GrabEdgeEvent<NInternalEvents::TEvSyncResponse>(edge, TDuration::Seconds(10)),
+            TEmptyEventQueueException
+        );
+    }
+
+    Y_UNIT_TEST(ReconfigurationWithCurrentSyncRequest) {
+        TTestBasicRuntime runtime;
+        SetupMinimalRuntime(runtime);
+
+        runtime.SetLogPriority(NKikimrServices::SCHEME_BOARD_SUBSCRIBER, NLog::PRI_DEBUG);
+
+        const auto stateStorageInfo = GetStateStorageInfo(runtime);
+        const auto participatingReplicas = CountParticipatingReplicas(*stateStorageInfo);
+
+        constexpr int DomainId = 1;
+        constexpr const char* Path = "TestPath";
+        const TActorId edge = runtime.AllocateEdgeActor();
+
+        const TActorId subscriber = runtime.Register(CreateSchemeBoardSubscriber(edge, Path, DomainId));
+        TBlockEvents<NInternalEvents::TEvNotify> notificationBlocker(runtime, [&](const NInternalEvents::TEvNotify::TPtr& ev) {
+            return ev->Recipient == subscriber;
+        });
+        runtime.WaitFor("initial path lookups", [&]() {
+            return notificationBlocker.size() == participatingReplicas;
+        }, TDuration::Seconds(10));
+        notificationBlocker.Stop().Unblock();
+
+        constexpr ui64 cookie = 12345;
+        TBlockEvents<NInternalEvents::TEvSyncVersionResponse> syncResponseBlocker(runtime, [&](const NInternalEvents::TEvSyncVersionResponse::TPtr& ev) {
+            return ev->Recipient == subscriber && ev->Cookie == cookie;
+        });
+        runtime.Send(new IEventHandle(subscriber, edge, new NInternalEvents::TEvSyncRequest(), 0, cookie));
+        runtime.WaitFor("some sync responses", [&]() {
+            return !syncResponseBlocker.empty();
+        }, TDuration::Seconds(10));
+        syncResponseBlocker.Unblock(1);
+
+        auto replicas = ResolveReplicas(runtime, Path);
+        runtime.Send(subscriber, edge, replicas->Release().Release());
+        syncResponseBlocker.Stop().Unblock();
+
+        auto syncResponse = runtime.GrabEdgeEvent<NInternalEvents::TEvSyncResponse>(edge, TDuration::Seconds(10));
+        UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Get()->Path, Path, syncResponse->ToString());
+        UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Cookie, cookie, syncResponse->ToString());
+        UNIT_ASSERT_VALUES_EQUAL_C(syncResponse->Get()->Partial, false, syncResponse->ToString());
+
+        // No additional sync responses.
+        UNIT_CHECK_GENERATED_EXCEPTION(
+            runtime.GrabEdgeEvent<NInternalEvents::TEvSyncResponse>(edge, TDuration::Seconds(10)),
+            TEmptyEventQueueException
+        );
+    }
 }
 
 } // NSchemeBoard

ydb/core/tx/scheme_board/ut_helpers.cpp

@@ -24,6 +24,16 @@ TIntrusiveConstPtr<TStateStorageInfo> GetStateStorageInfo(TTestActorRuntime& run
     return result->Get()->Info;
 }
 
+TEvStateStorage::TEvResolveReplicasList::TPtr ResolveReplicas(TTestActorRuntime& runtime, const TString& path) {
+    const TActorId recipient = MakeStateStorageProxyID();
+    const TActorId edge = runtime.AllocateEdgeActor();
+    runtime.Send(recipient, edge, new TEvStateStorage::TEvResolveSchemeBoard(path));
+
+    auto result = runtime.GrabEdgeEvent<TEvStateStorage::TEvResolveReplicasList>(edge);
+    UNIT_ASSERT(result);
+    return result;
+}
+
 NKikimrScheme::TEvDescribeSchemeResult GenerateDescribe(
     const TString& path,
     TPathId pathId,

ydb/core/tx/scheme_board/ut_helpers.h

@@ -5,6 +5,7 @@
 #include "events_internal.h"
 #include "subscriber.h"
 
+#include <ydb/core/base/statestorage_impl.h>
 #include <ydb/core/base/tablet_types.h>
 #include <ydb/core/protos/flat_tx_scheme.pb.h>
 #include <ydb/core/testlib/basics/appdata.h>
@@ -27,6 +28,7 @@ namespace NSchemeBoard {
 void SetupMinimalRuntime(TTestActorRuntime& runtime, const TStateStorageSetupper& setupStateStorage = CreateDefaultStateStorageSetupper());
 
 TIntrusiveConstPtr<TStateStorageInfo> GetStateStorageInfo(TTestActorRuntime& runtime);
+TEvStateStorage::TEvResolveReplicasList::TPtr ResolveReplicas(TTestActorRuntime& runtime, const TString& path);
 
 class TTestContext: public TTestBasicRuntime {
 public: