Retry NotifyBridgeSyncFinished command when racing

ydb/core/mind/bscontroller/bridge.cpp

@@ -73,15 +73,38 @@ class TBlobStorageController::TTxCheckUnsynced : public TTransactionBase<TBlobSt
         if (State) {
             State->ApplyConfigUpdates();
         }
+        Y_ABORT_UNLESS(Self->CheckingUnsyncedBridgePiles);
+        Self->CheckingUnsyncedBridgePiles = !SyncedPiles.empty();
+        Self->NotifyBridgeSyncFinishedErrors = false;
+        Self->NumPendingBridgeSyncFinishedResponses = SyncedPiles.size();
         for (TBridgePileId bridgePileId : SyncedPiles) {
             NKikimrBlobStorage::TEvNodeConfigInvokeOnRoot request;
             auto *cmd = request.MutableNotifyBridgeSyncFinished();
             cmd->SetGeneration(Config.GetClusterState().GetGeneration());
             bridgePileId.CopyToProto(cmd, &std::decay_t<decltype(*cmd)>::SetBridgePileId);
             cmd->SetBSC(true);
-            Self->InvokeOnRoot(std::move(request), [](auto&) {});
+
+            Self->InvokeOnRoot(std::move(request), [self = Self](NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult& result) {
+                if (result.GetStatus() != NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult::OK) {
+                    self->NotifyBridgeSyncFinishedErrors = true;
+                }
+                Y_ABORT_UNLESS(self->NumPendingBridgeSyncFinishedResponses);
+                if (!--self->NumPendingBridgeSyncFinishedResponses) {
+                    Y_ABORT_UNLESS(self->CheckingUnsyncedBridgePiles);
+                    self->CheckingUnsyncedBridgePiles = false;
+                    if (self->NotifyBridgeSyncFinishedErrors) {
+                        self->CheckUnsyncedBridgePiles();
+                    }
+                }
+            });
         }
         Self->ProcessSyncers();
+
+        // start checking piles again if config generation has changed
+        if (!Self->CheckingUnsyncedBridgePiles && Self->CheckingUnsyncedBridgePilesGeneration !=
+                Self->StorageConfig->GetGeneration()) {
+            Self->CheckUnsyncedBridgePiles();
+        }
     }
 };
 
@@ -224,14 +247,16 @@ class TBlobStorageController::TTxUpdateBridgeSyncState : public TTransactionBase
 };
 
 void TBlobStorageController::CheckUnsyncedBridgePiles() {
-    if (!StorageConfig->HasClusterStateDetails()) {
+    if (!StorageConfig->HasClusterStateDetails() || CheckingUnsyncedBridgePiles) {
         return; // no bridge mode cluster state details available
     }
     const auto& details = StorageConfig->GetClusterStateDetails();
     for (const auto& pss : details.GetPileSyncState()) {
         if (pss.GetUnsyncedBSC()) {
+            CheckingUnsyncedBridgePiles = true;
+            CheckingUnsyncedBridgePilesGeneration = StorageConfig->GetGeneration();
             Execute(std::make_unique<TTxCheckUnsynced>(this, *StorageConfig));
-            return;
+            break;
         }
     }
 }
@@ -419,8 +444,12 @@ void TBlobStorageController::ApplySyncerState(TNodeId nodeId, const NKikimrBlobS
                 groupId.CopyToProto(cmd, &std::decay_t<decltype(*cmd)>::SetGroupId);
                 cmd->SetGroupGeneration(generation);
                 cmd->MutableBridgeGroupInfo()->Swap(&bridgeGroupInfo);
-                InvokeOnRoot(std::move(request), [=](NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult& /*result*/) {
-                    // FIXME: check result
+                InvokeOnRoot(std::move(request), [=](NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult& result) {
+                    if (result.GetStatus() != NKikimrBlobStorage::TEvNodeConfigInvokeOnRootResult::OK) {
+                        Y_DEBUG_ABORT("UpdateBridgeGroupInfo has unexpectedly failed");
+                        STLOG(PRI_ERROR, BS_CONTROLLER, BSCBR08, "UpdateBridgeGroupInfo has unexpectedly failed",
+                            (Result, result));
+                    }
                     if (const auto it = TargetGroupToSyncerState.find(targetGroupId); it != TargetGroupToSyncerState.end()) {
                         TSyncerState& syncerState = it->second;
                         Y_ABORT_UNLESS(syncerState.InCommit);

ydb/core/mind/bscontroller/impl.h

@@ -2030,6 +2030,11 @@ class TBlobStorageController : public TActor<TBlobStorageController>, public TTa
     std::set<std::tuple<TNodeId, TSyncerState*>> NodeToSyncerState;
     TIntrusiveList<TSyncerState, TSyncersRequiringAction> SyncersRequiringAction;
 
+    ui64 CheckingUnsyncedBridgePilesGeneration = 0;
+    bool CheckingUnsyncedBridgePiles = false;
+    bool NotifyBridgeSyncFinishedErrors = false;
+    ui32 NumPendingBridgeSyncFinishedResponses = 0;
+
     void CheckUnsyncedBridgePiles();
 
     void ApplySyncerState(TNodeId nodeId, const NKikimrBlobStorage::TEvControllerUpdateSyncerState& update,