Propagate StorageConfig up the distconf tree (#25867)

Author: alexvru

ydb/core/blobstorage/nodewarden/distconf_binding.cpp

@@ -432,22 +432,37 @@ namespace NKikimr::NStorage {
             FanOutReversePush();
         }
 
+        std::unique_ptr<TEvNodeConfigPush> pendingPush;
+        auto getPendingPushRecord = [&] {
+            if (!pendingPush) {
+                pendingPush.reset(new TEvNodeConfigPush);
+            }
+            return &pendingPush->Record;
+        };
+
         // process cache updates, if needed
         if (record.HasCacheUpdate()) {
             auto *cacheUpdate = record.MutableCacheUpdate();
             ApplyCacheUpdates(cacheUpdate, senderNodeId);
 
             if (cacheUpdate->RequestedKeysSize()) {
-                auto ev = std::make_unique<TEvNodeConfigPush>();
                 for (const TString& key : cacheUpdate->GetRequestedKeys()) {
                     if (const auto it = Cache.find(key); it != Cache.end()) {
-                        AddCacheUpdate(ev->Record.MutableCacheUpdate(), it, true);
+                        AddCacheUpdate(getPendingPushRecord()->MutableCacheUpdate(), it, true);
                     }
                 }
-                SendEvent(*Binding, std::move(ev));
             }
         }
 
+        if (StorageConfig && record.HasRequestStorageConfigGeneration() &&
+                record.GetRequestStorageConfigGeneration() <= StorageConfig->GetGeneration()) {
+            getPendingPushRecord()->MutableStorageConfig()->CopyFrom(*StorageConfig);
+        }
+
+        if (pendingPush) {
+            SendEvent(*Binding, std::move(pendingPush));
+        }
+
         // if we have a root node id that is not ours, then we drop all inbound pile connections; also execute any
         // pending commands
         if (Binding && Binding->RootNodeId) {
@@ -537,11 +552,14 @@ namespace NKikimr::NStorage {
 
         // check if we have to send our current config to the peer
         const NKikimrBlobStorage::TStorageConfig *configToPeer = nullptr;
-        if (record.GetInitial()) {
+        std::optional<ui64> requestStorageConfigGeneration;
+        if (StorageConfig) {
             for (const auto& item : record.GetBoundNodes()) {
                 if (item.GetNodeId().GetNodeId() == senderNodeId) {
-                    if (StorageConfig && item.GetMeta().GetGeneration() < StorageConfig->GetGeneration()) {
+                    if (item.GetMeta().GetGeneration() < StorageConfig->GetGeneration()) {
                         configToPeer = StorageConfig.get();
+                    } else if (StorageConfig->GetGeneration() < item.GetMeta().GetGeneration()) {
+                        requestStorageConfigGeneration.emplace(item.GetMeta().GetGeneration());
                     }
                     break;
                 }
@@ -606,7 +624,8 @@ namespace NKikimr::NStorage {
             GetRootNodeId());
         TBoundNode& info = it->second;
         if (inserted) {
-            auto response = std::make_unique<TEvNodeConfigReversePush>(GetRootNodeId(), configToPeer);
+            auto response = std::make_unique<TEvNodeConfigReversePush>(GetRootNodeId(), configToPeer,
+                requestStorageConfigGeneration);
             if (record.GetInitial()) {
                 auto *cache = record.MutableCacheUpdate();
 
@@ -675,6 +694,10 @@ namespace NKikimr::NStorage {
             ApplyCacheUpdates(record.MutableCacheUpdate(), senderNodeId);
         }
 
+        if (record.HasStorageConfig()) {
+            ApplyStorageConfig(record.GetStorageConfig(), /*fromBinding=*/ false);
+        }
+
         if (pushEv && pushEv->IsUseful()) {
             SendEvent(*Binding, std::move(pushEv));
         }

ydb/core/blobstorage/nodewarden/distconf_cache.cpp

@@ -48,7 +48,7 @@ namespace NKikimr::NStorage {
         }
         // propagate backwards
         for (auto& [nodeId, info] : DirectBoundNodes) {
-            auto ev = std::make_unique<TEvNodeConfigReversePush>(GetRootNodeId(), nullptr);
+            auto ev = std::make_unique<TEvNodeConfigReversePush>(GetRootNodeId(), nullptr, std::nullopt);
             info.LastReportedRootNodeId = GetRootNodeId();
             ev->Record.MutableCacheUpdate()->CopyFrom(updates);
             SendEvent(nodeId, info, std::move(ev));

ydb/core/blobstorage/nodewarden/distconf_fsm.cpp

@@ -715,7 +715,7 @@ namespace NKikimr::NStorage {
             const bool needToUpdateConfig = nodeIdsToUpdate.contains(nodeId);
             if (needToUpdateConfig || info.LastReportedRootNodeId != rootNodeId) {
                 SendEvent(nodeId, info, std::make_unique<TEvNodeConfigReversePush>(rootNodeId,
-                    needToUpdateConfig ? StorageConfig.get() : nullptr));
+                    needToUpdateConfig ? StorageConfig.get() : nullptr, std::nullopt));
                 info.LastReportedRootNodeId = GetRootNodeId();
             }
         }

ydb/core/blobstorage/nodewarden/node_warden_events.h

@@ -22,11 +22,15 @@ namespace NKikimr::NStorage {
     {
         TEvNodeConfigReversePush() = default;
 
-        TEvNodeConfigReversePush(ui32 rootNodeId, const NKikimrBlobStorage::TStorageConfig *committedConfig) {
+        TEvNodeConfigReversePush(ui32 rootNodeId, const NKikimrBlobStorage::TStorageConfig *committedConfig,
+                std::optional<ui64> requestStorageConfigGeneration) {
             Record.SetRootNodeId(rootNodeId);
             if (committedConfig) {
                 Record.MutableCommittedStorageConfig()->CopyFrom(*committedConfig);
             }
+            if (requestStorageConfigGeneration) {
+                Record.SetRequestStorageConfigGeneration(*requestStorageConfigGeneration);
+            }
         }
 
         static std::unique_ptr<TEvNodeConfigReversePush> MakeRejected(const NKikimrBlobStorage::TStorageConfig *config) {

ydb/core/protos/blobstorage_distributed_config.proto

@@ -78,6 +78,7 @@ message TEvNodeConfigPush {
     repeated TBoundNode BoundNodes = 2; // a list of bound node updates (including itself)
     repeated TNodeIdentifier DeletedBoundNodeIds = 3; // a list of detached nodes
     TCacheUpdate CacheUpdate = 4;
+    TStorageConfig StorageConfig = 5;
 }
 
 // Used to reverse-propagate configuration and to confirm/reject initial TEvNodePushBinding query.
@@ -87,6 +88,7 @@ message TEvNodeConfigReversePush {
     TStorageConfig CommittedStorageConfig = 3; // last known committed storage configuration
     bool RecurseConfigUpdate = 4;
     TCacheUpdate CacheUpdate = 5;
+    optional uint64 RequestStorageConfigGeneration = 6;
 }
 
 // Remove node from bound list.