Support Console interoperation through distconf (#13109)

19 files changed, 526 insertions, 98 deletions

diff --git a/ydb/core/blobstorage/base/blobstorage_console_events.h b/ydb/core/blobstorage/base/blobstorage_console_events.h
index 4091bf2a8f0..6c3315145c3 100644
--- a/ydb/core/blobstorage/base/blobstorage_console_events.h
+++ b/ydb/core/blobstorage/base/blobstorage_console_events.h
@@ -10,7 +10,7 @@ namespace NKikimr {
         NKikimrBlobStorage::TEvControllerProposeConfigRequest, TEvBlobStorage::EvControllerProposeConfigRequest> {
         TEvControllerProposeConfigRequest() = default;
 
-        TEvControllerProposeConfigRequest(const ui32 configHash, const ui32 configVersion) {
+        TEvControllerProposeConfigRequest(ui64 configHash, ui32 configVersion) {
             Record.SetConfigHash(configHash);
             Record.SetConfigVersion(configVersion);
         }
@@ -68,6 +68,8 @@ namespace NKikimr {
     struct TEvBlobStorage::TEvControllerValidateConfigResponse : TEventPB<TEvBlobStorage::TEvControllerValidateConfigResponse,
         NKikimrBlobStorage::TEvControllerValidateConfigResponse, TEvBlobStorage::EvControllerValidateConfigResponse> {
         TEvControllerValidateConfigResponse() = default;
+
+        std::optional<TString> InternalError;
     };
 
     struct TEvBlobStorage::TEvControllerReplaceConfigRequest : TEventPB<TEvBlobStorage::TEvControllerReplaceConfigRequest,
diff --git a/ydb/core/blobstorage/nodewarden/distconf.cpp b/ydb/core/blobstorage/nodewarden/distconf.cpp
index f5361d45b38..dd507b5539b 100644
--- a/ydb/core/blobstorage/nodewarden/distconf.cpp
+++ b/ydb/core/blobstorage/nodewarden/distconf.cpp
@@ -1,6 +1,9 @@
 #include "distconf.h"
 #include "node_warden_impl.h"
 #include <ydb/core/mind/dynamic_nameserver.h>
+#include <ydb/library/yaml_config/yaml_config_helpers.h>
+#include <ydb/library/yaml_config/yaml_config.h>
+#include <library/cpp/streams/zstd/zstd.h>
 
 namespace NKikimr::NStorage {
 
@@ -61,16 +64,48 @@ namespace NKikimr::NStorage {
 
     bool TDistributedConfigKeeper::ApplyStorageConfig(const NKikimrBlobStorage::TStorageConfig& config) {
         if (!StorageConfig || StorageConfig->GetGeneration() < config.GetGeneration()) {
+            StorageConfigYaml = StorageConfigFetchYaml = {};
+            StorageConfigFetchYamlHash = 0;
+            StorageConfigYamlVersion.reset();
+
+            if (config.HasConfigComposite()) {
+                try {
+                    // parse the composite stream
+                    TStringInput ss(config.GetConfigComposite());
+                    TZstdDecompress zstd(&ss);
+                    StorageConfigYaml = TString::Uninitialized(LoadSize(&zstd));
+                    zstd.LoadOrFail(StorageConfigYaml.Detach(), StorageConfigYaml.size());
+                    StorageConfigFetchYaml = TString::Uninitialized(LoadSize(&zstd));
+                    zstd.LoadOrFail(StorageConfigFetchYaml.Detach(), StorageConfigFetchYaml.size());
+
+                    // extract _current_ config version
+                    auto metadata = NYamlConfig::GetMetadata(StorageConfigYaml);
+                    Y_DEBUG_ABORT_UNLESS(metadata.Version.has_value());
+                    StorageConfigYamlVersion = metadata.Version.value_or(0);
+
+                    // and _fetched_ config hash
+                    StorageConfigFetchYamlHash = NYaml::GetConfigHash(StorageConfigFetchYaml);
+                } catch (const std::exception& ex) {
+                    Y_ABORT("ConfigComposite format incorrect: %s", ex.what());
+                }
+            }
+
             StorageConfig.emplace(config);
             if (ProposedStorageConfig && ProposedStorageConfig->GetGeneration() <= StorageConfig->GetGeneration()) {
                 ProposedStorageConfig.reset();
             }
+
             Send(MakeBlobStorageNodeWardenID(SelfId().NodeId()), new TEvNodeWardenStorageConfig(*StorageConfig,
                 ProposedStorageConfig ? &ProposedStorageConfig.value() : nullptr));
+
             if (IsSelfStatic) {
                 PersistConfig({});
                 ApplyConfigUpdateToDynamicNodes(false);
             }
+
+            ConnectToConsole();
+            SendConfigProposeRequest();
+
             return true;
         } else if (StorageConfig->GetGeneration() && StorageConfig->GetGeneration() == config.GetGeneration() &&
                 StorageConfig->GetFingerprint() != config.GetFingerprint()) {
@@ -197,6 +232,10 @@ namespace NKikimr::NStorage {
             Y_ABORT_UNLESS(!Binding);
         } else {
             Y_ABORT_UNLESS(RootState == ERootState::INITIAL || RootState == ERootState::ERROR_TIMEOUT);
+
+            // we can't have connection to the Console without being the root node
+            Y_ABORT_UNLESS(!ConsolePipeId);
+            Y_ABORT_UNLESS(!ConsoleConnected);
         }
     }
 #endif
@@ -280,6 +319,11 @@ namespace NKikimr::NStorage {
             fFunc(TEvents::TSystem::Gone, HandleGone);
             cFunc(TEvents::TSystem::Wakeup, HandleWakeup);
             cFunc(TEvents::TSystem::Poison, PassAway);
+            hFunc(TEvTabletPipe::TEvClientConnected, Handle);
+            hFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
+            hFunc(TEvBlobStorage::TEvControllerValidateConfigResponse, Handle);
+            hFunc(TEvBlobStorage::TEvControllerProposeConfigResponse, Handle);
+            hFunc(TEvBlobStorage::TEvControllerConsoleCommitResponse, Handle);
         )
         for (ui32 nodeId : std::exchange(UnsubscribeQueue, {})) {
             UnsubscribeInterconnect(nodeId);
diff --git a/ydb/core/blobstorage/nodewarden/distconf.h b/ydb/core/blobstorage/nodewarden/distconf.h
index 79ebbd0f93d..a49bfc05195 100644
--- a/ydb/core/blobstorage/nodewarden/distconf.h
+++ b/ydb/core/blobstorage/nodewarden/distconf.h
@@ -177,6 +177,10 @@ namespace NKikimr::NStorage {
 
         // currently active storage config
         std::optional<NKikimrBlobStorage::TStorageConfig> StorageConfig;
+        TString StorageConfigYaml; // the part we have to push (unless this is storage-only) to console
+        TString StorageConfigFetchYaml; // the part we would get is we fetch from console
+        ui64 StorageConfigFetchYamlHash = 0;
+        std::optional<ui32> StorageConfigYamlVersion;
 
         // base config from config file
         NKikimrBlobStorage::TStorageConfig BaseConfig;
@@ -261,6 +265,17 @@ namespace NKikimr::NStorage {
         // child actors
         THashSet<TActorId> ChildActors;
 
+        // pipe to Console
+        TActorId ConsolePipeId;
+        bool ConsoleConnected = false;
+        bool ConfigCommittedToConsole = false;
+        ui64 ValidateRequestCookie = 0;
+        ui64 ProposeRequestCookie = 0;
+        ui64 CommitRequestCookie = 0;
+        bool ProposeRequestInFlight = false;
+        std::optional<std::tuple<ui64, ui32>> ProposedConfigHashVersion;
+        std::vector<std::tuple<TActorId, TString, ui64>> ConsoleConfigValidationQ;
+
         friend void ::Out<ERootState>(IOutputStream&, ERootState);
 
     public:
@@ -422,6 +437,23 @@ namespace NKikimr::NStorage {
         void Handle(NMon::TEvHttpInfo::TPtr ev);
 
         ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+        // Console interaction
+
+        void ConnectToConsole(bool enablingDistconf = false);
+        void DisconnectFromConsole();
+        void SendConfigProposeRequest();
+        void Handle(TEvBlobStorage::TEvControllerValidateConfigResponse::TPtr ev);
+        void Handle(TEvBlobStorage::TEvControllerProposeConfigResponse::TPtr ev);
+        void Handle(TEvBlobStorage::TEvControllerConsoleCommitResponse::TPtr ev);
+        void Handle(TEvTabletPipe::TEvClientConnected::TPtr ev);
+        void Handle(TEvTabletPipe::TEvClientDestroyed::TPtr ev);
+        void OnConsolePipeError();
+        bool EnqueueConsoleConfigValidation(TActorId queryId, bool enablingDistconf, TString yaml);
+
+        static std::optional<TString> UpdateConfigComposite(NKikimrBlobStorage::TStorageConfig& config, const TString& yaml,
+            const std::optional<TString>& fetched);
+
+        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
         // Consistency checking
 
 #ifdef NDEBUG
diff --git a/ydb/core/blobstorage/nodewarden/distconf_console.cpp b/ydb/core/blobstorage/nodewarden/distconf_console.cpp
new file mode 100644
index 00000000000..63e98dc7cab
--- /dev/null
+++ b/ydb/core/blobstorage/nodewarden/distconf_console.cpp
@@ -0,0 +1,244 @@
+#include "distconf.h"
+
+#include <ydb/library/yaml_config/yaml_config.h>
+#include <library/cpp/streams/zstd/zstd.h>
+
+namespace NKikimr::NStorage {
+
+    void TDistributedConfigKeeper::ConnectToConsole(bool enablingDistconf) {
+        if (ConsolePipeId) {
+            return; // connection is already established
+        } else if (!Scepter) {
+            return; // this is not the root node
+        } else if (enablingDistconf) {
+            // NO RETURN HERE -> right now we are enabling distconf, so we can skip rest of the checks
+        } else if (!StorageConfig || !StorageConfig->GetSelfManagementConfig().GetEnabled()) {
+            return; // no self-management config enabled
+        } else if (!StorageConfig->HasStateStorageConfig()) {
+            return; // no way to find Console too
+        }
+
+        STLOG(PRI_DEBUG, BS_NODE, NWDC66, "ConnectToConsole: creating pipe to the Console");
+        ConsolePipeId = Register(NTabletPipe::CreateClient(SelfId(), MakeConsoleID(),
+            NTabletPipe::TClientRetryPolicy::WithRetries()));
+    }
+
+    void TDistributedConfigKeeper::DisconnectFromConsole() {
+        NTabletPipe::CloseAndForgetClient(SelfId(), ConsolePipeId);
+        ConsoleConnected = false;
+    }
+
+    void TDistributedConfigKeeper::SendConfigProposeRequest() {
+        if (!ConsoleConnected) {
+            return;
+        }
+
+        if (ProposeRequestInFlight) {
+            return; // still waiting for previous one
+        }
+
+        if (!StorageConfig || !StorageConfig->HasConfigComposite()) {
+            return; // no config yet
+        }
+
+        Y_ABORT_UNLESS(StorageConfigYamlVersion);
+
+        STLOG(PRI_DEBUG, BS_NODE, NWDC67, "SendConfigProposeRequest: sending propose request to the Console",
+            (StorageConfigFetchYamlHash, StorageConfigFetchYamlHash),
+            (StorageConfigYamlVersion, StorageConfigYamlVersion),
+            (ProposedConfigHashVersion, ProposedConfigHashVersion),
+            (ProposeRequestCookie, ProposeRequestCookie + 1));
+
+        Y_DEBUG_ABORT_UNLESS(!ProposedConfigHashVersion || ProposedConfigHashVersion == std::make_tuple(
+            StorageConfigFetchYamlHash, *StorageConfigYamlVersion));
+        ProposedConfigHashVersion.emplace(StorageConfigFetchYamlHash, *StorageConfigYamlVersion);
+        NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerProposeConfigRequest(
+            StorageConfigFetchYamlHash, *StorageConfigYamlVersion), ++ProposeRequestCookie);
+        ProposeRequestInFlight = true;
+    }
+
+    void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerValidateConfigResponse::TPtr ev) {
+        auto& q = ConsoleConfigValidationQ;
+        auto pred = [&](const auto& item) {
+            const auto& [actorId, yaml, cookie] = item;
+            const bool match = cookie == ev->Cookie;
+            if (match) {
+                TActivationContext::Send(ev->Forward(actorId));
+            }
+            return match;
+        };
+        q.erase(std::remove_if(q.begin(), q.end(), pred), q.end());
+    }
+
+    void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerProposeConfigResponse::TPtr ev) {
+        STLOG(PRI_DEBUG, BS_NODE, NWDC68, "received TEvControllerProposeConfigResponse",
+            (ConsoleConnected, ConsoleConnected),
+            (ProposeRequestInFlight, ProposeRequestInFlight),
+            (Cookie, ev->Cookie),
+            (ProposeRequestCookie, ProposeRequestCookie),
+            (ProposedConfigHashVersion, ProposedConfigHashVersion),
+            (Record, ev->Get()->Record));
+
+        if (!ConsoleConnected || !ProposeRequestInFlight || ev->Cookie != ProposeRequestCookie) {
+            return;
+        }
+        ProposeRequestInFlight = false;
+
+        const auto& record = ev->Get()->Record;
+        switch (record.GetStatus()) {
+            case NKikimrBlobStorage::TEvControllerProposeConfigResponse::HashMismatch:
+            case NKikimrBlobStorage::TEvControllerProposeConfigResponse::UnexpectedConfig:
+                // TODO: error condition; restart?
+                ProposedConfigHashVersion.reset();
+                break;
+
+            case NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNeeded: {
+                if (!StorageConfig || !StorageConfig->HasConfigComposite() || ProposedConfigHashVersion !=
+                        std::make_tuple(StorageConfigFetchYamlHash, *StorageConfigYamlVersion)) {
+                    const char *err = "proposed config, but something has gone awfully wrong";
+                    STLOG(PRI_CRIT, BS_NODE, NWDC69, err, (StorageConfig, StorageConfig),
+                        (ProposedConfigHashVersion, ProposedConfigHashVersion),
+                        (StorageConfigFetchYamlHash, StorageConfigFetchYamlHash),
+                        (StorageConfigYamlVersion, StorageConfigYamlVersion));
+                    Y_DEBUG_ABORT("%s", err);
+                    return;
+                }
+
+                NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerConsoleCommitRequest(
+                    StorageConfigYaml), ++CommitRequestCookie);
+                break;
+            }
+
+            case NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNotNeeded:
+                // it's okay, just wait for another configuration change or something like that
+                ConfigCommittedToConsole = true;
+                break;
+        }
+    }
+
+    void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerConsoleCommitResponse::TPtr ev) {
+        STLOG(PRI_DEBUG, BS_NODE, NWDC70, "received TEvControllerConsoleCommitResponse",
+            (ConsoleConnected, ConsoleConnected),
+            (Cookie, ev->Cookie),
+            (CommitRequestCookie, CommitRequestCookie),
+            (Record, ev->Get()->Record));
+
+        if (!ConsoleConnected || ev->Cookie != CommitRequestCookie) {
+            return;
+        }
+
+        const auto& record = ev->Get()->Record;
+        switch (record.GetStatus()) {
+            case NKikimrBlobStorage::TEvControllerConsoleCommitResponse::SessionMismatch:
+                DisconnectFromConsole();
+                ConnectToConsole();
+                break;
+
+            case NKikimrBlobStorage::TEvControllerConsoleCommitResponse::NotCommitted:
+                break;
+
+            case NKikimrBlobStorage::TEvControllerConsoleCommitResponse::Committed:
+                ConfigCommittedToConsole = true;
+                break;
+        }
+
+        ProposedConfigHashVersion.reset();
+    }
+
+    void TDistributedConfigKeeper::Handle(TEvTabletPipe::TEvClientConnected::TPtr ev) {
+        STLOG(PRI_DEBUG, BS_NODE, NWDC71, "received TEvClientConnected", (ConsolePipeId, ConsolePipeId),
+            (TabletId, ev->Get()->TabletId), (Status, ev->Get()->Status), (ClientId, ev->Get()->ClientId),
+            (ServerId, ev->Get()->ServerId));
+        if (ev->Get()->ClientId == ConsolePipeId) {
+            if (ev->Get()->Status == NKikimrProto::OK) {
+                Y_ABORT_UNLESS(!ConsoleConnected);
+                ConsoleConnected = true;
+                SendConfigProposeRequest();
+                for (auto& [actorId, yaml, cookie] : ConsoleConfigValidationQ) {
+                    Y_ABORT_UNLESS(!cookie);
+                    cookie = ++ValidateRequestCookie;
+                    NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerValidateConfigRequest(
+                        yaml), cookie);
+                }
+            } else {
+                OnConsolePipeError();
+            }
+        }
+    }
+
+    void TDistributedConfigKeeper::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr ev) {
+        STLOG(PRI_DEBUG, BS_NODE, NWDC72, "received TEvClientDestroyed", (ConsolePipeId, ConsolePipeId),
+            (TabletId, ev->Get()->TabletId), (ClientId, ev->Get()->ClientId), (ServerId, ev->Get()->ServerId));
+        if (ev->Get()->ClientId == ConsolePipeId) {
+            OnConsolePipeError();
+        }
+    }
+
+    void TDistributedConfigKeeper::OnConsolePipeError() {
+        ConsolePipeId = {};
+        ConsoleConnected = false;
+        ConfigCommittedToConsole = false;
+        ProposedConfigHashVersion.reset();
+        ProposeRequestInFlight = false;
+        ++CommitRequestCookie; // to prevent processing any messages
+
+        // cancel any pending requests
+        for (const auto& [actorId, yaml, cookie] : ConsoleConfigValidationQ) {
+            auto ev = std::make_unique<TEvBlobStorage::TEvControllerValidateConfigResponse>();
+            ev->InternalError = "pipe disconnected";
+            Send(actorId, ev.release());
+        }
+        ConsoleConfigValidationQ.clear();
+
+        ConnectToConsole();
+    }
+
+    std::optional<TString> TDistributedConfigKeeper::UpdateConfigComposite(NKikimrBlobStorage::TStorageConfig& config,
+            const TString& yaml, const std::optional<TString>& fetched) {
+        TString temp;
+        const TString *finalFetchedConfig = fetched ? &fetched.value() : &temp;
+
+        if (!fetched) { // fill in 'to-be-fetched' version of config with version incremented by one
+            try {
+                auto metadata = NYamlConfig::GetMetadata(yaml);
+                metadata.Cluster = metadata.Cluster.value_or("unknown"); // TODO: fix this
+                metadata.Version = metadata.Version.value_or(0) + 1;
+                temp = NYamlConfig::ReplaceMetadata(yaml, metadata);
+            } catch (const std::exception& ex) {
+                return ex.what();
+            }
+        }
+
+        TStringStream ss;
+        {
+            TZstdCompress zstd(&ss);
+            SaveSize(&zstd, yaml.size());
+            zstd.Write(yaml);
+            SaveSize(&zstd, finalFetchedConfig->size());
+            zstd.Write(*finalFetchedConfig);
+        }
+        config.SetConfigComposite(ss.Str());
+
+        return {};
+    }
+
+    bool TDistributedConfigKeeper::EnqueueConsoleConfigValidation(TActorId actorId, bool enablingDistconf, TString yaml) {
+        if (!ConsolePipeId) {
+            ConnectToConsole(enablingDistconf);
+            if (!ConsolePipeId) {
+                return false;
+            }
+        }
+
+        auto& [qActorId, qYaml, qCookie] = ConsoleConfigValidationQ.emplace_back(actorId, std::move(yaml), 0);
+
+        if (ConsoleConnected) {
+            qCookie = ++ValidateRequestCookie;
+            NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerValidateConfigRequest(qYaml),
+                qCookie);
+        }
+
+        return true;
+    }
+
+}
diff --git a/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp b/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp
index 160ac71cb92..af967ae8f8f 100644
--- a/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp
+++ b/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp
@@ -40,9 +40,14 @@ namespace NKikimr::NStorage {
         task.SetTaskId(RandomNumber<ui64>());
         task.MutableCollectConfigs();
         IssueScatterTask(TActorId(), std::move(task));
+
+        // establish connection to console tablet (if we have means to do it)
+        Y_ABORT_UNLESS(!ConsolePipeId);
+        ConnectToConsole();
     }
 
     void TDistributedConfigKeeper::UnbecomeRoot() {
+        DisconnectFromConsole();
     }
 
     void TDistributedConfigKeeper::SwitchToError(const TString& reason) {
@@ -349,8 +354,7 @@ namespace NKikimr::NStorage {
                 configToPropose = persistedConfig;
             }
         } else if (baseConfig && !baseConfig->GetGeneration()) {
-            const bool canBootstrapAutomatically = baseConfig->HasSelfManagementConfig() &&
-                baseConfig->GetSelfManagementConfig().GetEnabled() &&
+            const bool canBootstrapAutomatically = baseConfig->GetSelfManagementConfig().GetEnabled() &&
                 baseConfig->GetSelfManagementConfig().GetAutomaticBootstrap();
             if (canBootstrapAutomatically || selfAssemblyUUID) {
                 if (!selfAssemblyUUID) {
diff --git a/ydb/core/blobstorage/nodewarden/distconf_generate.cpp b/ydb/core/blobstorage/nodewarden/distconf_generate.cpp
index dc9330a9fa5..8a47b3d1a09 100644
--- a/ydb/core/blobstorage/nodewarden/distconf_generate.cpp
+++ b/ydb/core/blobstorage/nodewarden/distconf_generate.cpp
@@ -2,13 +2,15 @@
 
 #include <ydb/core/mind/bscontroller/group_geometry_info.h>
 
+#include <ydb/library/yaml_config/yaml_config_helpers.h>
+#include <ydb/library/yaml_json/yaml_to_json.h>
 #include <library/cpp/streams/zstd/zstd.h>
 
 namespace NKikimr::NStorage {
 
     std::optional<TString> TDistributedConfigKeeper::GenerateFirstConfig(NKikimrBlobStorage::TStorageConfig *config,
             const TString& selfAssemblyUUID) {
-        if (!config->HasSelfManagementConfig() || !config->GetSelfManagementConfig().GetEnabled()) {
+        if (!config->GetSelfManagementConfig().GetEnabled()) {
             return "self-management is not enabled";
         }
         const auto& smConfig = config->GetSelfManagementConfig();
@@ -45,11 +47,24 @@ namespace NKikimr::NStorage {
             return "missing initial config YAML";
         }
         TStringStream ss;
-        {
-            TZstdCompress zstd(&ss);
-            zstd << Cfg->SelfManagementConfig->GetInitialConfigYaml();
+        TString yaml = Cfg->SelfManagementConfig->GetInitialConfigYaml();
+        ui32 version = 0;
+        try {
+            auto json = NYaml::Yaml2Json(YAML::Load(yaml), true);
+            if (json.Has("metadata")) {
+                if (auto& metadata = json["metadata"]; metadata.Has("version")) {
+                    version = metadata["version"].GetUIntegerRobust();
+                }
+            }
+        } catch (const std::exception& ex) {
+            return TStringBuilder() << "failed to parse initial config YAML: " << ex.what();
+        }
+        if (version) {
+            return TStringBuilder() << "initial config version must be zero";
+        }
+        if (const auto& error = UpdateConfigComposite(*config, yaml, std::nullopt)) {
+            return TStringBuilder() << "failed to update config yaml: " << *error;
         }
-        config->SetStorageConfigCompressedYAML(ss.Str());
 
         if (!Cfg->DomainsConfig) { // no automatic configuration required
         } else if (Cfg->DomainsConfig->StateStorageSize() == 1) { // the StateStorage config is already defined explicitly, just migrate it
diff --git a/ydb/core/blobstorage/nodewarden/distconf_invoke.cpp b/ydb/core/blobstorage/nodewarden/distconf_invoke.cpp
index 8d62ee3a58e..d98c2f6a4d0 100644
--- a/ydb/core/blobstorage/nodewarden/distconf_invoke.cpp
+++ b/ydb/core/blobstorage/nodewarden/distconf_invoke.cpp
@@ -2,7 +2,7 @@
 #include "node_warden_impl.h"
 
 #include <ydb/library/yaml_config/yaml_config_parser.h>
-#include <library/cpp/streams/zstd/zstd.h>
+#include <ydb/library/yaml_json/yaml_to_json.h>
 
 namespace NKikimr::NStorage {
 
@@ -25,6 +25,12 @@ namespace NKikimr::NStorage {
         ui64 NextScatterCookie = 1;
         THashMap<ui64, TGatherCallback> ScatterTasks;
 
+        std::shared_ptr<TLifetimeToken> RequestHandlerToken = std::make_shared<TLifetimeToken>();
+
+        NKikimrBlobStorage::TStorageConfig ProposedStorageConfig;
+
+        TString NewYaml;
+
     public:
         TInvokeRequestHandlerActor(TDistributedConfigKeeper *self, std::unique_ptr<TEventHandle<TEvNodeConfigInvokeOnRoot>>&& ev)
             : Self(self)
@@ -108,23 +114,8 @@ namespace NKikimr::NStorage {
             auto& record = Event->Get()->Record;
             STLOG(PRI_DEBUG, BS_NODE, NWDC43, "ExecuteQuery", (SelfId, SelfId()), (Record, record));
             switch (record.GetRequestCase()) {
-                case TQuery::kUpdateConfig: {
-                    auto *request = record.MutableUpdateConfig();
-
-                    if (!RunCommonChecks()) {
-                        return;
-                    }
-
-                    auto *config = request->MutableConfig();
-
-                    if (auto error = ValidateConfig(*Self->StorageConfig)) {
-                        return FinishWithError(TResult::ERROR, TStringBuilder() << "current config validation failed: " << *error);
-                    } else if (auto error = ValidateConfigUpdate(*Self->StorageConfig, *config)) {
-                        return FinishWithError(TResult::ERROR, TStringBuilder() << "config validation failed: " << *error);
-                    }
-
-                    return StartProposition(config);
-                }
+                case TQuery::kUpdateConfig:
+                    return UpdateConfig(record.MutableUpdateConfig());
 
                 case TQuery::kQueryConfig: {
                     auto ev = PrepareResult(TResult::OK, std::nullopt);
@@ -158,7 +149,7 @@ namespace NKikimr::NStorage {
                     return FetchStorageConfig(record.GetFetchStorageConfig().GetManual());
 
                 case TQuery::kReplaceStorageConfig:
-                    return ReplaceStorageConfig(record.GetReplaceStorageConfig().GetYAML());
+                    return ReplaceStorageConfig(record.GetReplaceStorageConfig());
 
                 case TQuery::kBootstrapCluster:
                     return BootstrapCluster(record.GetBootstrapCluster().GetSelfAssemblyUUID());
@@ -198,6 +189,25 @@ namespace NKikimr::NStorage {
         }
 
         ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+        // Configuration update
+
+        void UpdateConfig(TQuery::TUpdateConfig *request) {
+            if (!RunCommonChecks()) {
+                return;
+            }
+
+            auto *config = request->MutableConfig();
+
+            if (auto error = ValidateConfig(*Self->StorageConfig)) {
+                return FinishWithError(TResult::ERROR, TStringBuilder() << "UpdateConfig current config validation failed: " << *error);
+            } else if (auto error = ValidateConfigUpdate(*Self->StorageConfig, *config)) {
+                return FinishWithError(TResult::ERROR, TStringBuilder() << "UpdateConfig config validation failed: " << *error);
+            }
+
+            StartProposition(config);
+       }
+
+        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
         // Reassign group disk logic
 
         void ReassignGroupDisk(const TQuery::TReassignGroupDisk& cmd) {
@@ -249,7 +259,7 @@ namespace NKikimr::NStorage {
                 const TActorId actorId = GroupInfo->GetActorId(i);
                 const ui32 flags = IEventHandle::FlagTrackDelivery |
                     (actorId.NodeId() == SelfId().NodeId() ? 0 : IEventHandle::FlagSubscribeOnSession);
-                STLOG(PRI_DEBUG, BS_NODE, NW53, "sending TEvVStatus", (SelfId, SelfId()), (VDiskId, vdiskId),
+                STLOG(PRI_DEBUG, BS_NODE, NWDC73, "sending TEvVStatus", (SelfId, SelfId()), (VDiskId, vdiskId),
                     (ActorId, actorId));
                 Send(actorId, new TEvBlobStorage::TEvVStatus(vdiskId), flags);
                 if (actorId.NodeId() != SelfId().NodeId()) {
@@ -263,7 +273,7 @@ namespace NKikimr::NStorage {
         void Handle(TEvBlobStorage::TEvVStatusResult::TPtr ev) {
             const auto& record = ev->Get()->Record;
             const TVDiskID vdiskId = VDiskIDFromVDiskID(record.GetVDiskID());
-            STLOG(PRI_DEBUG, BS_NODE, NW54, "TEvVStatusResult", (SelfId, SelfId()), (Record, record), (VDiskId, vdiskId));
+            STLOG(PRI_DEBUG, BS_NODE, NWDC74, "TEvVStatusResult", (SelfId, SelfId()), (Record, record), (VDiskId, vdiskId));
             if (!PendingVDiskIds.erase(vdiskId)) {
                 return FinishWithError(TResult::ERROR, TStringBuilder() << "TEvVStatusResult VDiskID# " << vdiskId
                     << " is unexpected");
@@ -307,7 +317,7 @@ namespace NKikimr::NStorage {
                 return;
             }
 
-            STLOG(PRI_DEBUG, BS_NODE, NW55, "ReassignGroupDiskExecute", (SelfId, SelfId()));
+            STLOG(PRI_DEBUG, BS_NODE, NWDC75, "ReassignGroupDiskExecute", (SelfId, SelfId()));
 
             const auto& vdiskId = VDiskIDFromVDiskID(cmd.GetVDiskId());
 
@@ -373,7 +383,7 @@ namespace NKikimr::NStorage {
             }
             const auto& ss = bsConfig.GetServiceSet();
 
-            if (!config.HasSelfManagementConfig() || !config.GetSelfManagementConfig().GetEnabled()) {
+            if (!config.GetSelfManagementConfig().GetEnabled()) {
                 return FinishWithError(TResult::ERROR, "self-management is not enabled");
             }
             const auto& smConfig = config.GetSelfManagementConfig();
@@ -411,7 +421,7 @@ namespace NKikimr::NStorage {
                             &BaseConfig.value(), cmd.GetConvertToDonor(), cmd.GetIgnoreVSlotQuotaCheck(),
                             cmd.GetIsSelfHealReasonDecommit());
                     } catch (const TExConfigError& ex) {
-                        STLOG(PRI_NOTICE, BS_NODE, NW49, "ReassignGroupDisk failed to allocate group", (SelfId, SelfId()),
+                        STLOG(PRI_NOTICE, BS_NODE, NWDC76, "ReassignGroupDisk failed to allocate group", (SelfId, SelfId()),
                             (Config, config),
                             (BaseConfig, *BaseConfig),
                             (Error, ex.what()));
@@ -628,35 +638,49 @@ namespace NKikimr::NStorage {
         void FetchStorageConfig(bool manual) {
             if (!Self->StorageConfig) {
                 FinishWithError(TResult::ERROR, "no agreed StorageConfig");
-            } else if (!Self->StorageConfig->HasStorageConfigCompressedYAML()) {
+            } else if (!Self->StorageConfigFetchYaml) {
                 FinishWithError(TResult::ERROR, "no stored YAML for storage config");
             } else {
                 auto ev = PrepareResult(TResult::OK, std::nullopt);
                 auto *record = &ev->Record;
-                TStringInput ss(Self->StorageConfig->GetStorageConfigCompressedYAML());
-                TString config = TZstdDecompress(&ss).ReadAll();
+                record->MutableFetchStorageConfig()->SetYAML(Self->StorageConfigFetchYaml);
 
                 if (manual) {
                     // add BlobStorageConfig, NameserviceConfig, DomainsConfig
                 }
 
-                record->MutableFetchStorageConfig()->SetYAML(config);
                 Finish(Sender, SelfId(), ev.release(), 0, Cookie);
             }
         }
 
-        void ReplaceStorageConfig(const TString& yaml) {
+        void ReplaceStorageConfig(const TQuery::TReplaceStorageConfig& request) {
             if (!RunCommonChecks()) {
                 return;
+            } else if (!Self->ConfigCommittedToConsole && Self->StorageConfig->GetSelfManagementConfig().GetEnabled()) {
+                return FinishWithError(TResult::ERROR, "previous config has not been committed to Console yet");
             }
 
+            NewYaml = request.GetYAML();
+            ui32 newYamlVersion = 0;
+
             NKikimrConfig::TAppConfig appConfig;
             try {
-                appConfig = NKikimr::NYaml::Parse(yaml);
+                auto json = NYaml::Yaml2Json(YAML::Load(NewYaml), true);
+                NYaml::Parse(json, NYaml::GetJsonToProtoConfig(), appConfig, true);
+                if (json.Has("metadata")) {
+                    if (auto& metadata = json["metadata"]; metadata.Has("version")) {
+                        newYamlVersion = metadata["version"].GetUIntegerRobust();
+                    }
+                }
             } catch (const std::exception& ex) {
                 return FinishWithError(TResult::ERROR, TStringBuilder() << "exception while parsing YAML: " << ex.what());
             }
 
+            if (Self->StorageConfigYamlVersion && newYamlVersion != *Self->StorageConfigYamlVersion + 1) {
+                return FinishWithError(TResult::ERROR, TStringBuilder() << "version must be increasing by one"
+                    << " new version# " << newYamlVersion << " expected version# " << *Self->StorageConfigYamlVersion);
+            }
+
             TString errorReason;
             NKikimrBlobStorage::TStorageConfig config(*Self->StorageConfig);
             const bool success = DeriveStorageConfig(appConfig, &config, &errorReason);
@@ -665,15 +689,64 @@ namespace NKikimr::NStorage {
                     << errorReason);
             }
 
-            TStringStream ss;
-            {
-                TZstdCompress zstd(&ss);
-                zstd << yaml;
+            if (const auto& error = UpdateConfigComposite(config, NewYaml, std::nullopt)) {
+                return FinishWithError(TResult::ERROR, TStringBuilder() << "failed to update config yaml: " << *error);
             }
-            config.SetStorageConfigCompressedYAML(ss.Str());
 
+            // advance the config generation
             config.SetGeneration(config.GetGeneration() + 1);
-            StartProposition(&config);
+
+            if (auto error = ValidateConfig(*Self->StorageConfig)) {
+                return FinishWithError(TResult::ERROR, TStringBuilder()
+                    << "ReplaceStorageConfig current config validation failed: " << *error);
+            } else if (auto error = ValidateConfigUpdate(*Self->StorageConfig, config)) {
+                return FinishWithError(TResult::ERROR, TStringBuilder()
+                    << "ReplaceStorageConfig config validation failed: " << *error);
+            }
+
+            const bool pushToConsole = true;
+
+            if (!pushToConsole || !request.GetSkipConsoleValidation()) {
+                return StartProposition(&config);
+            }
+
+            // whether we are enabling distconf right now
+            const bool enablingDistconf = Self->StorageConfig->GetSelfManagementConfig().GetEnabled() <
+                config.GetSelfManagementConfig().GetEnabled();
+
+            if (!Self->EnqueueConsoleConfigValidation(SelfId(), enablingDistconf, NewYaml)) {
+                FinishWithError(TResult::ERROR, "console pipe is not available");
+            } else {
+                ProposedStorageConfig = std::move(config);
+            }
+        }
+
+        void Handle(TEvBlobStorage::TEvControllerValidateConfigResponse::TPtr ev) {
+            const auto& record = ev->Get()->Record;
+            STLOG(PRI_DEBUG, BS_NODE, NWDC77, "received TEvControllerValidateConfigResponse", (SelfId, SelfId()),
+                (InternalError, ev->Get()->InternalError), (Status, record.GetStatus()));
+
+            if (ev->Get()->InternalError) {
+                return FinishWithError(TResult::ERROR, TStringBuilder() << "failed to validate config through console: "
+                    << *ev->Get()->InternalError);
+            }
+
+            switch (record.GetStatus()) {
+                case NKikimrBlobStorage::TEvControllerValidateConfigResponse::IdPipeServerMismatch:
+                    Self->DisconnectFromConsole();
+                    Self->ConnectToConsole();
+                    return FinishWithError(TResult::ERROR, TStringBuilder() << "console connection race detected");
+
+                case NKikimrBlobStorage::TEvControllerValidateConfigResponse::ConfigNotValid:
+                    return FinishWithError(TResult::ERROR, TStringBuilder() << "console config validation failed: "
+                        << record.GetErrorReason());
+
+                case NKikimrBlobStorage::TEvControllerValidateConfigResponse::ConfigIsValid:
+                    if (const auto& error = UpdateConfigComposite(ProposedStorageConfig, NewYaml, record.GetYAML())) {
+                        return FinishWithError(TResult::ERROR, TStringBuilder() << "failed to update config yaml: " << *error);
+                    }
+                    return StartProposition(&ProposedStorageConfig);
+            }
         }
 
         void BootstrapCluster(const TString& selfAssemblyUUID) {
@@ -748,9 +821,10 @@ namespace NKikimr::NStorage {
             }
 
             if (auto error = ValidateConfigUpdate(*Self->StorageConfig, *config)) {
-                STLOG(PRI_DEBUG, BS_NODE, NW51, "proposed config validation failed", (SelfId, SelfId()), (Error, *error),
-                    (Config, config));
-                return FinishWithError(TResult::ERROR, TStringBuilder() << "config validation failed: " << *error);
+                STLOG(PRI_DEBUG, BS_NODE, NWDC78, "StartProposition config validation failed", (SelfId, SelfId()),
+                    (Error, *error), (Config, config));
+                return FinishWithError(TResult::ERROR, TStringBuilder()
+                    << "StartProposition config validation failed: " << *error);
             }
 
             Self->CurrentProposedStorageConfig.emplace(std::move(*config));
@@ -844,6 +918,7 @@ namespace NKikimr::NStorage {
                 hFunc(TEvents::TEvUndelivered, Handle);
                 hFunc(TEvNodeWardenBaseConfig, Handle);
                 cFunc(TEvents::TSystem::Poison, PassAway);
+                hFunc(TEvBlobStorage::TEvControllerValidateConfigResponse, Handle);
             )
         }
     };
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_impl.cpp b/ydb/core/blobstorage/nodewarden/node_warden_impl.cpp
index 2b38b422fed..45a756ecafb 100644
--- a/ydb/core/blobstorage/nodewarden/node_warden_impl.cpp
+++ b/ydb/core/blobstorage/nodewarden/node_warden_impl.cpp
@@ -1071,7 +1071,7 @@ bool NKikimr::NStorage::DeriveStorageConfig(const NKikimrConfig::TAppConfig& app
         };
 
         // update static group information unless distconf is enabled
-        if (!hasStaticGroupInfo(ssFrom) && config->HasSelfManagementConfig() && config->GetSelfManagementConfig().GetEnabled()) {
+        if (!hasStaticGroupInfo(ssFrom) && config->GetSelfManagementConfig().GetEnabled()) {
             // distconf enabled, keep it as is
         } else if (!hasStaticGroupInfo(*ssTo)) {
             ssTo->MutablePDisks()->CopyFrom(ssFrom.GetPDisks());
diff --git a/ydb/core/blobstorage/nodewarden/ya.make b/ydb/core/blobstorage/nodewarden/ya.make
index 0afc988b0cb..9647b50e6fd 100644
--- a/ydb/core/blobstorage/nodewarden/ya.make
+++ b/ydb/core/blobstorage/nodewarden/ya.make
@@ -6,6 +6,7 @@ SRCS(
     distconf.cpp
     distconf.h
     distconf_binding.cpp
+    distconf_console.cpp
     distconf_dynamic.cpp
     distconf_generate.cpp
     distconf_fsm.cpp
diff --git a/ydb/core/cms/console/console_handshake.cpp b/ydb/core/cms/console/console_handshake.cpp
index f26fddf4377..43605b5ad39 100644
--- a/ydb/core/cms/console/console_handshake.cpp
+++ b/ydb/core/cms/console/console_handshake.cpp
@@ -8,6 +8,7 @@
 
 #include <ydb/core/base/tablet_pipe.h>
 #include <ydb/core/util/stlog.h>
+#include <ydb/core/util/pb.h>
 
 #include <ydb/library/actors/core/hfunc.h>
 
@@ -15,10 +16,11 @@ namespace NKikimr::NConsole {
 
 class TConfigsManager::TConsoleCommitActor : public TActorBootstrapped<TConsoleCommitActor> {
 public:
-    TConsoleCommitActor(TActorId senderId, const TString& yamlConfig, TActorId interconnectSession)
+    TConsoleCommitActor(TActorId senderId, const TString& yamlConfig, TActorId interconnectSession, ui64 cookie)
         : SenderId(senderId)
         , YamlConfig(yamlConfig)
         , InterconnectSession(interconnectSession)
+        , Cookie(cookie)
     {}
 
     void Bootstrap(const TActorId& consoleId) {
@@ -32,14 +34,15 @@ public:
     void Handle(TEvConsole::TEvReplaceYamlConfigResponse::TPtr& /* ev */) {
         auto response = std::make_unique<TEvBlobStorage::TEvControllerConsoleCommitResponse>();
         response->Record.SetStatus(NKikimrBlobStorage::TEvControllerConsoleCommitResponse::Committed);
-        SendInReply(SenderId, InterconnectSession, std::move(response));
+        SendInReply(std::move(response));
         PassAway();
     }
 
-    void Handle(TEvConsole::TEvGenericError::TPtr& /* ev */) {
+    void Handle(TEvConsole::TEvGenericError::TPtr& ev) {
         auto response = std::make_unique<TEvBlobStorage::TEvControllerConsoleCommitResponse>();
         response->Record.SetStatus(NKikimrBlobStorage::TEvControllerConsoleCommitResponse::NotCommitted);
-        SendInReply(SenderId, InterconnectSession, std::move(response));
+        response->Record.SetErrorReason(SingleLineProto(ev->Get()->Record));
+        SendInReply(std::move(response));
         PassAway();
     }
 
@@ -53,11 +56,12 @@ private:
     TActorId SenderId;
     TString YamlConfig;
     TActorId InterconnectSession;
+    ui64 Cookie;
 
-    void SendInReply(const TActorId& senderId, const TActorId& interconnectSession, std::unique_ptr<IEventBase> ev) {
-        auto h = std::make_unique<IEventHandle>(senderId, SelfId(), ev.release(), 0, 0);
-        if (interconnectSession) {
-            h->Rewrite(TEvInterconnect::EvForward, interconnectSession);
+    void SendInReply(std::unique_ptr<IEventBase> ev) {
+        auto h = std::make_unique<IEventHandle>(SenderId, SelfId(), ev.release(), 0, Cookie);
+        if (InterconnectSession) {
+            h->Rewrite(TEvInterconnect::EvForward, InterconnectSession);
         }
         TActivationContext::Send(h.release());
     }
@@ -89,29 +93,22 @@ void TConfigsManager::Handle(TEvBlobStorage::TEvControllerProposeConfigRequest::
     auto response = std::make_unique<TEvBlobStorage::TEvControllerProposeConfigResponse>();
     auto& responseRecord = response->Record;
 
-    if (YamlVersion > record.GetConfigVersion()) {
+    if (YamlVersion == proposedConfigVersion) {
+        responseRecord.SetStatus(NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNeeded);
+    } else if (YamlVersion != proposedConfigVersion && (proposedConfigVersion && YamlVersion != proposedConfigVersion - 1)) {
         responseRecord.SetStatus(NKikimrBlobStorage::TEvControllerProposeConfigResponse::UnexpectedConfig);
         responseRecord.SetProposedConfigVersion(proposedConfigVersion);
         responseRecord.SetConsoleConfigVersion(YamlVersion);
-        SendInReply(ev->Sender, ev->InterconnectSession, std::move(response));
         LOG_ALERT_S(ctx, NKikimrServices::CMS, "Unexpected proposed config.");
-        return;
-    }
-    if (YamlVersion == record.GetConfigVersion()) {
-        if (proposedConfigHash != currentConfigHash) {
-            responseRecord.SetStatus(NKikimrBlobStorage::TEvControllerProposeConfigResponse::HashMismatch);
-            responseRecord.SetProposedConfigHash(proposedConfigHash);
-            responseRecord.SetConsoleConfigHash(currentConfigHash);
-            SendInReply(ev->Sender, ev->InterconnectSession, std::move(response));
-            LOG_ALERT_S(ctx, NKikimrServices::CMS, "Config hash mismatch.");
-            return;
-        }
+    } else if (proposedConfigHash != currentConfigHash) {
+        responseRecord.SetStatus(NKikimrBlobStorage::TEvControllerProposeConfigResponse::HashMismatch);
+        responseRecord.SetProposedConfigHash(proposedConfigHash);
+        responseRecord.SetConsoleConfigHash(currentConfigHash);
+        LOG_ALERT_S(ctx, NKikimrServices::CMS, "Config hash mismatch.");
+    } else {
         responseRecord.SetStatus(NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNotNeeded);
-        SendInReply(ev->Sender, ev->InterconnectSession, std::move(response));
-        return;
     }
-    responseRecord.SetStatus(NKikimrBlobStorage::TEvControllerProposeConfigResponse::CommitIsNeeded);
-    SendInReply(ev->Sender, ev->InterconnectSession, std::move(response));
+    SendInReply(ev->Sender, ev->InterconnectSession, std::move(response), ev->Cookie);
 }
 
 void TConfigsManager::Handle(TEvBlobStorage::TEvControllerConsoleCommitRequest::TPtr& ev, const TActorContext& /* ctx */) {
@@ -121,7 +118,7 @@ void TConfigsManager::Handle(TEvBlobStorage::TEvControllerConsoleCommitRequest::
         return;
     }
 
-    IActor* actor = new TConsoleCommitActor(ev->Sender, yamlConfig, ev->InterconnectSession);
+    IActor* actor = new TConsoleCommitActor(ev->Sender, yamlConfig, ev->InterconnectSession, ev->Cookie);
     CommitActor = Register(actor);
 }
 
@@ -130,7 +127,6 @@ void TConfigsManager::Handle(TEvBlobStorage::TEvControllerValidateConfigRequest:
     auto requestRecord = ev->Get()->Record;
     response->Record.SetSkipBSCValidation(requestRecord.GetSkipBSCValidation());
     response->Record.SetConfigVersion(requestRecord.GetConfigVersion());
-    response->Record.SetYAML(requestRecord.GetYAML());
     auto& record = response->Record;
     auto yamlConfig = requestRecord.GetYAML();
     if (!CheckSession(*ev, response, NKikimrBlobStorage::TEvControllerValidateConfigResponse::IdPipeServerMismatch)) {
@@ -139,13 +135,19 @@ void TConfigsManager::Handle(TEvBlobStorage::TEvControllerValidateConfigRequest:
     auto result = ValidateConfigAndReplaceMetadata(yamlConfig);
     if (result.ErrorReason || result.HasForbiddenUnknown) {
         record.SetStatus(NKikimrBlobStorage::TEvControllerValidateConfigResponse::ConfigNotValid);
-        SendInReply(ev->Sender, ev->InterconnectSession, std::move(response));
-        return;
+        if (!result.ErrorReason) {
+            record.SetErrorReason("has forbidden unknown fields");
+        } else {
+            record.SetErrorReason(*result.ErrorReason);
+            if (result.HasForbiddenUnknown) {
+                record.SetErrorReason(record.GetErrorReason() + " + has forbidden unknown fields");
+            }
+        }
+    } else {
+        record.SetStatus(NKikimrBlobStorage::TEvControllerValidateConfigResponse::ConfigIsValid);
+        record.SetYAML(result.UpdatedConfig);
     }
-
-    record.SetStatus(NKikimrBlobStorage::TEvControllerValidateConfigResponse::ConfigIsValid);
-    record.SetSkipBSCValidation(ev->Get()->Record.GetSkipBSCValidation());
-    SendInReply(ev->Sender, ev->InterconnectSession, std::move(response));
+    SendInReply(ev->Sender, ev->InterconnectSession, std::move(response), ev->Cookie);
 }
 
 }
diff --git a/ydb/core/config/init/init.cpp b/ydb/core/config/init/init.cpp
index 0f88193302e..28d5b06bb37 100644
--- a/ydb/core/config/init/init.cpp
+++ b/ydb/core/config/init/init.cpp
@@ -589,7 +589,7 @@ void LoadYamlConfig(TConfigRefs refs, const TString& yamlConfigFile, NKikimrConf
 
     const TString yamlConfigString = protoConfigFileProvider.GetProtoFromFile(yamlConfigFile, errorCollector);
 
-    if (appConfig.HasSelfManagementConfig() && appConfig.GetSelfManagementConfig().GetEnabled()) {
+    if (appConfig.GetSelfManagementConfig().GetEnabled()) {
         // fill in InitialConfigYaml only when self-management through distconf is enabled
         appConfig.MutableSelfManagementConfig()->SetInitialConfigYaml(yamlConfigString);
     }
diff --git a/ydb/core/grpc_services/rpc_bsconfig.cpp b/ydb/core/grpc_services/rpc_bsconfig.cpp
index c9aeb5af522..fba182f3136 100644
--- a/ydb/core/grpc_services/rpc_bsconfig.cpp
+++ b/ydb/core/grpc_services/rpc_bsconfig.cpp
@@ -109,7 +109,7 @@ public:
         } catch (const std::exception&) {
             return false; // assuming no distconf enabled in this config
         }
-        return newConfig.HasSelfManagementConfig() && newConfig.GetSelfManagementConfig().GetEnabled();
+        return newConfig.GetSelfManagementConfig().GetEnabled();
     }
 };
 
diff --git a/ydb/core/mind/bscontroller/bsc.cpp b/ydb/core/mind/bscontroller/bsc.cpp
index 744c0eaf640..0dc98ec2b3d 100644
--- a/ydb/core/mind/bscontroller/bsc.cpp
+++ b/ydb/core/mind/bscontroller/bsc.cpp
@@ -158,13 +158,16 @@ void TBlobStorageController::Handle(TEvNodeWardenStorageConfig::TPtr ev) {
         }
     }
 
-    if (StorageConfig.HasSelfManagementConfig() && StorageConfig.GetSelfManagementConfig().GetEnabled()) {
+    if (StorageConfig.GetSelfManagementConfig().GetEnabled()) {
         // assuming that in autoconfig mode HostRecords are managed by the distconf; we need to apply it here to
         // avoid race with box autoconfiguration and node list change
         HostRecords = std::make_shared<THostRecordMap::element_type>(StorageConfig);
         if (SelfHealId) {
             Send(SelfHealId, new TEvPrivate::TEvUpdateHostRecords(HostRecords));
         }
+    } else {
+        StartConsoleInteraction();
+        ConsoleInteraction->Start(); // start console interaction when working in non-distconf mode
     }
 
     if (!std::exchange(StorageConfigObtained, true)) { // this is the first time we get StorageConfig in this instance of BSC
@@ -190,8 +193,7 @@ void TBlobStorageController::Handle(TEvents::TEvUndelivered::TPtr ev) {
 
 void TBlobStorageController::ApplyStorageConfig(bool ignoreDistconf) {
     if (!StorageConfig.HasBlobStorageConfig() || // this would be strange
-            !ignoreDistconf && (!StorageConfig.HasSelfManagementConfig() ||
-            !StorageConfig.GetSelfManagementConfig().GetEnabled() ||
+            !ignoreDistconf && (!StorageConfig.GetSelfManagementConfig().GetEnabled() ||
             !StorageConfig.GetSelfManagementConfig().GetAutomaticBoxManagement())) {
         return;
     }
diff --git a/ydb/core/mind/bscontroller/console_interaction.cpp b/ydb/core/mind/bscontroller/console_interaction.cpp
index e21cfadcede..8ae0ccba933 100644
--- a/ydb/core/mind/bscontroller/console_interaction.cpp
+++ b/ydb/core/mind/bscontroller/console_interaction.cpp
@@ -9,7 +9,6 @@ namespace NKikimr::NBsController {
 
     void TBlobStorageController::StartConsoleInteraction() {
         ConsoleInteraction = std::make_unique<TConsoleInteraction>(*this);
-        ConsoleInteraction->Start();
         STLOG(PRI_DEBUG, BS_CONTROLLER, BSC22, "Console interaction started");
     }
 
@@ -112,7 +111,6 @@ namespace NKikimr::NBsController {
         }
     }
 
-
     void TBlobStorageController::TConsoleInteraction::Handle(TEvBlobStorage::TEvControllerConsoleCommitResponse::TPtr &ev) {
         STLOG(PRI_DEBUG, BS_CONTROLLER, BSC20, "Console commit config response", (Response, ev->Get()->Record));
         auto& record = ev->Get()->Record;
@@ -149,7 +147,8 @@ namespace NKikimr::NBsController {
         GRPCSenderId = ev->Sender;
         if (!record.GetSkipConsoleValidation()) {
             if (!ValidationTimeout) {
-                TActivationContext::Schedule(TDuration::Minutes(2), new IEventHandle(Self.SelfId(), Self.SelfId(), new TEvPrivate::TEvValidationTimeout()));
+                TActivationContext::Schedule(TDuration::Minutes(2), new IEventHandle(Self.SelfId(), Self.SelfId(),
+                    new TEvPrivate::TEvValidationTimeout()));
             }
             ValidationTimeout = TActivationContext::Now() + TDuration::Minutes(2);
             auto validateConfigEv = std::make_unique<TEvBlobStorage::TEvControllerValidateConfigRequest>();
@@ -163,7 +162,8 @@ namespace NKikimr::NBsController {
         Self.Send(ev->Sender, validateConfigResponse.release());
     }
 
-    bool TBlobStorageController::TConsoleInteraction::ParseConfig(const TString& config, ui32& /* configVersion */, NKikimrBlobStorage::TStorageConfig& storageConfig) {
+    bool TBlobStorageController::TConsoleInteraction::ParseConfig(const TString& config, ui32& /*configVersion*/,
+            NKikimrBlobStorage::TStorageConfig& storageConfig) {
         NKikimrConfig::TAppConfig appConfig;
         try {
             appConfig = NKikimr::NYaml::Parse(config);
@@ -174,8 +174,9 @@ namespace NKikimr::NBsController {
         return success;
     }
 
-    TBlobStorageController::TConsoleInteraction::TCommitConfigResult::EStatus TBlobStorageController::TConsoleInteraction::CheckConfig(const TString& config, ui32& configVersion, bool skipBSCValidation,
-                                                                                                                                       NKikimrBlobStorage::TStorageConfig& storageConfig) {
+    TBlobStorageController::TConsoleInteraction::TCommitConfigResult::EStatus
+            TBlobStorageController::TConsoleInteraction::CheckConfig(const TString& config, ui32& configVersion,
+            bool skipBSCValidation, NKikimrBlobStorage::TStorageConfig& storageConfig) {
         if (!ParseConfig(config, configVersion, storageConfig)) {
             return TConsoleInteraction::TCommitConfigResult::EStatus::ParseError;
         }
@@ -221,7 +222,8 @@ namespace NKikimr::NBsController {
         NKikimrBlobStorage::TStorageConfig StorageConfig;
         TString yamlConfig = record.GetYAML();
         ui32 configVersion = record.GetConfigVersion();
-        if (CheckConfig(yamlConfig, configVersion, record.GetSkipBSCValidation(), StorageConfig) == TConsoleInteraction::TCommitConfigResult::EStatus::Success) {
+        if (CheckConfig(yamlConfig, configVersion, record.GetSkipBSCValidation(), StorageConfig) ==
+                TConsoleInteraction::TCommitConfigResult::EStatus::Success) {
             Self.Execute(Self.CreateTxCommitConfig(yamlConfig, configVersion, StorageConfig));
             return;
         }
diff --git a/ydb/core/mind/bscontroller/impl.h b/ydb/core/mind/bscontroller/impl.h
index 3cd8c0ca69a..2ed7c7e0359 100644
--- a/ydb/core/mind/bscontroller/impl.h
+++ b/ydb/core/mind/bscontroller/impl.h
@@ -2098,7 +2098,10 @@ public:
         SignalTabletActive(TActivationContext::AsActorContext());
         Loaded = true;
         ApplyStorageConfig();
-        StartConsoleInteraction();
+
+        if (!ConsoleInteraction) { // maybe we are in distconf mode, no console interaction yet started
+            StartConsoleInteraction();
+        }
 
         for (const auto& [id, info] : GroupMap) {
             if (info->VirtualGroupState) {
diff --git a/ydb/core/mind/bscontroller/self_heal.cpp b/ydb/core/mind/bscontroller/self_heal.cpp
index f35210e0f6a..bfb57b1a66f 100644
--- a/ydb/core/mind/bscontroller/self_heal.cpp
+++ b/ydb/core/mind/bscontroller/self_heal.cpp
@@ -956,7 +956,7 @@ namespace NKikimr::NBsController {
 
     void TBlobStorageController::PushStaticGroupsToSelfHeal() {
         if (!SelfHealId || !StorageConfigObtained || !StorageConfig.HasBlobStorageConfig() ||
-                !StorageConfig.HasSelfManagementConfig() || !StorageConfig.GetSelfManagementConfig().GetEnabled()) {
+                !StorageConfig.GetSelfManagementConfig().GetEnabled()) {
             return;
         }
 
diff --git a/ydb/core/mind/dynamic_nameserver.cpp b/ydb/core/mind/dynamic_nameserver.cpp
index cc6ccce004a..c46b4ff3c2f 100644
--- a/ydb/core/mind/dynamic_nameserver.cpp
+++ b/ydb/core/mind/dynamic_nameserver.cpp
@@ -146,7 +146,7 @@ void TDynamicNameserver::Bootstrap(const TActorContext &ctx)
 void TDynamicNameserver::Handle(TEvNodeWardenStorageConfig::TPtr ev) {
     Y_ABORT_UNLESS(ev->Get()->Config);
     const auto& config = *ev->Get()->Config;
-    if (config.HasSelfManagementConfig() && config.GetSelfManagementConfig().GetEnabled()) {
+    if (config.GetSelfManagementConfig().GetEnabled()) {
         // self-management through distconf is enabled and we are operating based on their tables, so apply them now
         auto newStaticConfig = BuildNameserverTable(config);
         if (StaticConfig->StaticNodeTable != newStaticConfig->StaticNodeTable) {
diff --git a/ydb/core/protos/blobstorage.proto b/ydb/core/protos/blobstorage.proto
index b84e4f3bfcb..735480a96af 100644
--- a/ydb/core/protos/blobstorage.proto
+++ b/ydb/core/protos/blobstorage.proto
@@ -1392,6 +1392,7 @@ message TEvControllerConsoleCommitResponse {
         Committed = 2;
     }
     optional EStatus Status = 1;
+    optional string ErrorReason = 2;
 }
 
 message TEvControllerReplaceConfigRequest {
@@ -1430,6 +1431,7 @@ message TEvControllerValidateConfigResponse {
     optional bool SkipBSCValidation = 2;
     optional string YAML = 3;
     optional uint32 ConfigVersion = 4;
+    optional string ErrorReason = 5;
 }
 
 message TEvControllerValidationTimeout {
diff --git a/ydb/core/protos/blobstorage_distributed_config.proto b/ydb/core/protos/blobstorage_distributed_config.proto
index b4aa96dd227..a27e02fb9b8 100644
--- a/ydb/core/protos/blobstorage_distributed_config.proto
+++ b/ydb/core/protos/blobstorage_distributed_config.proto
@@ -28,9 +28,8 @@ message TStorageConfig { // contents of storage metadata
     repeated TNodeIdentifier AllNodes = 5; // set of all known nodes
     string ClusterUUID = 6; // cluster GUID as provided in nameservice config
     string SelfAssemblyUUID = 7; // self-assembly UUID generated when config is first created
-    optional bytes StorageConfigCompressedYAML = 12; // compressed stored storage config YAML (if stored)
+    optional bytes ConfigComposite = 12; // compressed stored storage config YAML (if stored)
     NKikimrConfig.TSelfManagementConfig SelfManagementConfig = 13;
-
     TStorageConfig PrevConfig = 100; // previous version of StorageConfig (if any)
 }
 
@@ -196,6 +195,7 @@ message TEvNodeConfigInvokeOnRoot {
 
     message TReplaceStorageConfig {
         string YAML = 1;
+        bool SkipConsoleValidation = 2;
     }
 
     message TBootstrapCluster {