NBS-6784: Аdd host to dbg (#40841)

40 files changed, 1382 insertions, 17 deletions

diff --git a/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service.h b/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service.h
index 424aecef9f3..ab49eda44fe 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service.h
@@ -38,6 +38,8 @@ struct IPartitionDirectService
     virtual void UpdateVChunkConfig(
         const NStorage::NPartitionDirect::TVChunkConfig& cfg) = 0;
 
+    virtual void RequestAddHost(size_t directBlockGroupId) = 0;
+
     // Generates the next tablet-wide write LSN. Called by a vchunk on its
     // executor thread when it starts processing a write, so generation and
     // dirty-map registration happen on the same thread. Also drives periodic
diff --git a/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service_mock.h b/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service_mock.h
index 97c609548e1..47f0320c1c3 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service_mock.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service_mock.h
@@ -4,6 +4,8 @@
 
 #include <ydb/core/nbs/cloud/storage/core/libs/coroutine/executor.h>
 
+#include <util/generic/vector.h>
+
 namespace NYdb::NBS::NBlockStore {
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -16,6 +18,8 @@ struct TPartitionDirectServiceMock: public IPartitionDirectService
 
     TVolumeConfigPtr VolumeConfig;
     bool DropScheduledCallbacks = false;
+    TVector<size_t> AddHostRequests;
+    ui64 LsnGenerator = 0;
 
     [[nodiscard]] TVolumeConfigPtr GetVolumeConfig() const override
     {
@@ -46,7 +50,10 @@ struct TPartitionDirectServiceMock: public IPartitionDirectService
         Y_UNUSED(cfg);
     }
 
-    ui64 LsnGenerator = 0;
+    void RequestAddHost(size_t directBlockGroupId) override
+    {
+        AddHostRequests.push_back(directBlockGroupId);
+    }
 
     ui64 GenerateLsn() override
     {
diff --git a/ydb/core/nbs/cloud/blockstore/libs/service/storage_test.h b/ydb/core/nbs/cloud/blockstore/libs/service/storage_test.h
index b4973d053ab..1c53cd0b488 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/service/storage_test.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/service/storage_test.h
@@ -47,6 +47,11 @@ public:
         Y_UNUSED(cfg);
     }
 
+    void RequestAddHost(size_t directBlockGroupId) override
+    {
+        Y_UNUSED(directBlockGroupId);
+    }
+
     ui64 LsnGenerator = 0;
 
     ui64 GenerateLsn() override
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group.h
index 306fbaffd8c..63b41cd15e9 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group.h
@@ -210,6 +210,14 @@ public:
 
     // Query dump for DirectBlockGroup and VChunks.
     virtual NThreading::TFuture<TDBGDumpResponse> Dump() = 0;
+
+    // Result of the DBG's AddHost request. On success (empty error) applies the
+    // new host; on failure (e.g. rejected at MaxHostCount) logs the reason.
+    virtual void OnAddHostResult(
+        const NProto::TError& error,
+        THostIndex newHostIndex,
+        NKikimrBlobStorage::NDDisk::TDDiskId ddiskId,
+        NKikimrBlobStorage::NDDisk::TDDiskId pbufferId) = 0;
 };
 
 using IDirectBlockGroupPtr = std::shared_ptr<IDirectBlockGroup>;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.cpp
index 66fea1c07c4..9650a75bafe 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.cpp
@@ -114,8 +114,8 @@ TDirectBlockGroup::TDirectBlockGroup(
               .Generation = TabletGeneration})
     , Oracle(StorageConfig, this)
 {
-    Y_ASSERT(pbufferIds.size() == DirectBlockGroupHostCount);
-    Y_ASSERT(ddisksIds.size() == DirectBlockGroupHostCount);
+    Y_ASSERT(pbufferIds.size() >= DirectBlockGroupHostCount);
+    Y_ASSERT(ddisksIds.size() >= DirectBlockGroupHostCount);
 
     auto addDDiskConnections = [&](const TVector<NBsController::TDDiskId>& ids,
                                    TVector<TDDiskConnection>& connections,
@@ -153,11 +153,20 @@ TDirectBlockGroup::TDirectBlockGroup(
         pbufferIds,
         PBufferConnections,
         EConnectionType::PBuffer);
+
+    GrowOracleToConnections();
 }
 
-void TDirectBlockGroup::Register(TVChunkWeakPtr vChunk)
+void TDirectBlockGroup::Register(TVChunkWeakPtr weakVChunk)
 {
-    VChunks.push_back(std::move(vChunk));
+    Y_ABORT_UNLESS(ExecutorThreadChecker.Check());
+
+    // Catch a vchunk up as it registers: its config can lag the connections
+    // after an add-host that committed just before a restart.
+    if (auto vChunk = weakVChunk.lock()) {
+        GrowVChunkToConnections(*vChunk);
+    }
+    VChunks.push_back(std::move(weakVChunk));
 }
 
 TExecutorPtr TDirectBlockGroup::GetExecutor()
@@ -1115,6 +1124,122 @@ void TDirectBlockGroup::SetHostState(
     }
 }
 
+void TDirectBlockGroup::QueryAddHost()
+{
+    Y_ABORT_UNLESS(ExecutorThreadChecker.Check());
+    Y_ABORT_UNLESS(Service);
+
+    // No gate here: the authoritative MaxHostCount check is in the partition
+    // (the DBG's DDiskConnections count lags). The DBG just forwards.
+    LOG_INFO(
+        *ActorSystem,
+        NKikimrServices::NBS_PARTITION,
+        "%s QueryAddHost dbgIndex=%zu",
+        LogTitle.GetWithTime().c_str(),
+        DirectBlockGroupIndex);
+
+    Service->RequestAddHost(DirectBlockGroupIndex);
+}
+
+void TDirectBlockGroup::OnAddHostResult(
+    const NProto::TError& error,
+    THostIndex newHostIndex,
+    NKikimrBlobStorage::NDDisk::TDDiskId ddiskId,
+    NKikimrBlobStorage::NDDisk::TDDiskId pbufferId)
+{
+    Y_ABORT_UNLESS(ExecutorThreadChecker.Check());
+
+    if (HasError(error)) {
+        LOG_WARN(
+            *ActorSystem,
+            NKikimrServices::NBS_PARTITION,
+            "%s AddHost request did not go through: %s",
+            LogTitle.GetWithTime().c_str(),
+            FormatError(error).c_str());
+        return;
+    }
+
+    Y_ABORT_UNLESS(
+        static_cast<size_t>(newHostIndex) == DDiskConnections.size(),
+        "AddHost expects appending at the end (newHostIndex %lu vs size %lu)",
+        static_cast<size_t>(newHostIndex),
+        DDiskConnections.size());
+    Y_ABORT_UNLESS(DDiskConnections.size() == PBufferConnections.size());
+    Y_ABORT_UNLESS(DDiskConnections.size() < MaxHostCount);
+    Y_ABORT_UNLESS(!DDiskConnections.empty());
+
+    LOG_INFO(
+        *ActorSystem,
+        NKikimrServices::NBS_PARTITION,
+        "%s AddHost newHostIndex=%s",
+        LogTitle.GetWithTime().c_str(),
+        PrintHostIndex(newHostIndex).c_str());
+
+    const ui32 generation =
+        DDiskConnections.front().HostConnection.Credentials.Generation;
+
+    NBsController::TDDiskId ddiskIdNative(ddiskId);
+    NBsController::TDDiskId pbufferIdNative(pbufferId);
+
+    DDiskConnections.push_back(TDDiskConnection{
+        .HostConnection = NTransport::THostConnection{
+            .ConnectionType = EConnectionType::DDisk,
+            .DDiskId = ddiskIdNative,
+            .Credentials = NDDisk::TQueryCredentials::ToDDisk(
+                TabletId,
+                generation,
+                InitialDDiskSessionSeqNo,
+                std::nullopt)}});
+
+    PBufferConnections.push_back(TDDiskConnection{
+        .HostConnection = NTransport::THostConnection{
+            .ConnectionType = EConnectionType::PBuffer,
+            .DDiskId = pbufferIdNative,
+            .Credentials = NDDisk::TQueryCredentials::ToPersistentBuffer(
+                TabletId,
+                generation,
+                std::nullopt)}});
+
+    Y_ABORT_UNLESS(
+        PBufferIdToHostIndex.insert({pbufferId, newHostIndex}).second);
+
+    SyncHostsWithConnections();
+
+    DoEstablishConnection(newHostIndex, EConnectionType::DDisk);
+    DoEstablishConnection(newHostIndex, EConnectionType::PBuffer);
+}
+
+void TDirectBlockGroup::SyncHostsWithConnections()
+{
+    Y_ABORT_UNLESS(ExecutorThreadChecker.Check());
+
+    for (const auto& weakVChunk: VChunks) {
+        if (auto vChunk = weakVChunk.lock()) {
+            GrowVChunkToConnections(*vChunk);
+        }
+    }
+    GrowOracleToConnections();
+}
+
+void TDirectBlockGroup::GrowVChunkToConnections(TVChunk& vChunk)
+{
+    // Idempotent: a vchunk already at the connection count gets nothing.
+    const size_t targetHostCount = DDiskConnections.size();
+    for (size_t hostCount = vChunk.GetConfig().GetHostCount() + 1;
+         hostCount <= targetHostCount;
+         ++hostCount)
+    {
+        vChunk.OnHostAppended(hostCount);
+    }
+}
+
+void TDirectBlockGroup::GrowOracleToConnections()
+{
+    while (Oracle.GetHostCount() < DDiskConnections.size()) {
+        Oracle.OnHostAdded();
+    }
+}
+
 ui64 TDirectBlockGroup::GetHostPBufferUsedSize(THostIndex hostIndex) const
 {
     ui64 result = 0;
@@ -1250,6 +1375,14 @@ void TDirectBlockGroup::OnConnectionEstablished(
             Oracle.OnDDiskConnected(index, TInstant::Now());
         }
         // INVARIANT: PBuffer does NOT require a session/lock
+    } else if (IsInitialized()) {
+        LOG_ERROR(
+            *ActorSystem,
+            NKikimrServices::NBS_PARTITION,
+            "%s connection failed for host %s (post-init): %s",
+            LogTitle.GetWithTime().c_str(),
+            PrintHostIndex(static_cast<THostIndex>(index)).c_str(),
+            FormatError(error).c_str());
     } else {
         // TODO (future phase): handle the error code/BLOCKED, transition to
         // Broken/suicide.
@@ -1259,9 +1392,7 @@ void TDirectBlockGroup::OnConnectionEstablished(
     // ConnectPromise resolves both "connection ready" and "session ready" in
     // this phase. Unblocks waiters in ReadFromDDisk/WriteToDDisk/ListPBuffers.
     connection.ConnectPromise.SetValue(error);
-    if (!InitialReadyPromise.HasValue() && HasLockedQuorum() &&
-        HasPBufferQuorum())
-    {
+    if (!IsInitialized() && HasLockedQuorum() && HasPBufferQuorum()) {
         InitialReadyPromise.SetValue();
         LOG_INFO(
             *ActorSystem,
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.h
index d81a5877476..2eeccf39e5a 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl.h
@@ -126,7 +126,11 @@ public:
 
     NThreading::TFuture<TDBGDumpResponse> Dump() override;
 
-    ui64 GetDDiskSessionSeqNo(size_t index) const;
+    void OnAddHostResult(
+        const NProto::TError& error,
+        THostIndex newHostIndex,
+        NKikimrBlobStorage::NDDisk::TDDiskId ddiskId,
+        NKikimrBlobStorage::NDDisk::TDDiskId pbufferId) override;
 
     // IHostStateController implementation
     void SetHostState(
@@ -134,6 +138,10 @@ public:
         EHostState oldState,
         EHostState newState) override;
     ui64 GetHostPBufferUsedSize(THostIndex hostIndex) const override;
+    void QueryAddHost() override;
+
+    // Own methods (not part of any interface).
+    ui64 GetDDiskSessionSeqNo(size_t index) const;
 
 private:
     using TEvSyncResult = NKikimrBlobStorage::NDDisk::TEvSyncResult;
@@ -169,6 +177,14 @@ private:
 
     void DoEstablishConnections();
     void DoEstablishConnection(size_t index, EConnectionType connectionType);
+
+    // Live AddHost: grow all vchunks and the Oracle to the new connection.
+    void SyncHostsWithConnections();
+
+    // Catch one vchunk / the Oracle up to the current connection count.
+    void GrowVChunkToConnections(TVChunk& vChunk);
+    void GrowOracleToConnections();
+
     void OnConnectionEstablished(
         EConnectionType connectionType,
         size_t index,
@@ -180,6 +196,11 @@ private:
     [[nodiscard]] bool HasPBufferQuorum() const;
     [[nodiscard]] bool HasLockedQuorum() const;
 
+    [[nodiscard]] bool IsInitialized() const
+    {
+        return InitialReadyPromise.HasValue();
+    }
+
     void DoListPBuffers();
     void OnPBuffersListed(const TAggregatedListPBufferResponse& response);
 
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl_ut.cpp
index 84ea861b4ce..e4bc275746e 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_impl_ut.cpp
@@ -1,4 +1,5 @@
 #include "direct_block_group_test_fixture.h"
+#include "vchunk.h"
 
 #include <ydb/core/nbs/cloud/blockstore/libs/common/constants.h>
 #include <ydb/core/nbs/cloud/blockstore/libs/service/partition_direct_service_mock.h>
@@ -28,9 +29,20 @@ using EConnectionType = NTransport::THostConnection::EConnectionType;
 using TStorageTransportMock = NTransport::TStorageTransportMock;
 using TICStorageTransportTestAdapter =
     NTransport::NTestLib::TICStorageTransportTestAdapter;
+using TDDiskId = NBsController::TDDiskId;
 
 ////////////////////////////////////////////////////////////////////////////////
 
+TVector<TDDiskId> MakeDDiskIds(ui32 baseNodeId, ui32 count)
+{
+    TVector<TDDiskId> ids;
+    ids.reserve(count);
+    for (ui32 i = 0; i < count; ++i) {
+        ids.emplace_back(baseNodeId + i, 1, i);
+    }
+    return ids;
+}
+
 TGuardedSgList MakeSgList(TString& buffer)
 {
     return TGuardedSgList(TSgList{TBlockDataRef{buffer.data(), buffer.size()}});
@@ -687,6 +699,100 @@ Y_UNIT_TEST_SUITE(TDirectBlockGroupTest)
             UNIT_ASSERT_VALUES_EQUAL(0, errorsInfo.ConsecutiveErrorCount);
         }
     }
+
+    // QueryAddHost() routes an add-host request to the partition-direct
+    // service, tagged with this DBG's index.
+    Y_UNIT_TEST_F(ShouldQueryAddHostThroughService, TDBGFixture)
+    {
+        auto executor = MakeExecutor();
+        auto dbg = MakeDirectBlockGroup(
+            executor,
+            std::make_unique<TStorageTransportMock>());
+
+        auto initialReady = RunAndGetInitialReady(dbg);
+        WaitReady(executor, initialReady);
+
+        auto& service = *Services.back();
+        UNIT_ASSERT_VALUES_EQUAL(0u, service.AddHostRequests.size());
+
+        RunOnExecutor(
+            executor,
+            [&]
+            {
+                dbg->QueryAddHost();
+                return true;
+            })
+            .GetValue(WaitTimeout);
+
+        UNIT_ASSERT_VALUES_EQUAL(1u, service.AddHostRequests.size());
+        UNIT_ASSERT_VALUES_EQUAL(
+            static_cast<size_t>(0),
+            service.AddHostRequests[0]);
+    }
+
+    // On restart a DBG comes up with the committed connection count (here N+1).
+    // The Oracle is born at that count in the constructor, and a vchunk whose
+    // persisted config still lags at N is caught up by the DBG the moment it
+    // registers - the same OnHostAppended path a live AddHost uses, with no
+    // partition involved.
+    Y_UNIT_TEST_F(ShouldCatchUpHostsOnStartup, TDBGFixture)
+    {
+        constexpr ui32 grownHostCount = DirectBlockGroupHostCount + 1;
+        constexpr ui64 vChunkSize = RegionSize / DirectBlockGroupsCount;
+
+        auto executor = MakeExecutor();
+
+        // The DBG comes up already grown to N+1 connections.
+        auto dbg = MakeDirectBlockGroup(
+            executor,
+            std::make_unique<TStorageTransportMock>(),
+            MakeDDiskIds(100, grownHostCount),
+            MakeDDiskIds(100 + grownHostCount, grownHostCount));
+
+        auto initialReady = RunAndGetInitialReady(dbg);
+        WaitReady(executor, initialReady);
+
+        // A vchunk that still only knows the pre-add host count.
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> counters(
+            new ::NMonitoring::TDynamicCounters());
+        auto vchunk = std::make_shared<TVChunk>(
+            Runtime->GetActorSystem(0),
+            Services.back().get(),
+            TVChunkConfig::MakeDefault(
+                100,
+                DirectBlockGroupHostCount,
+                DefaultPrimaryCount),
+            dbg,
+            3,
+            vChunkSize,
+            counters);
+
+        TString oracleDump;
+        TString dumpBefore;
+        TString dumpAfter;
+        RunOnExecutor(
+            executor,
+            [&]
+            {
+                oracleDump = dbg->GetOracle()->Dump();
+                dumpBefore = vchunk->DebugPrintDirtyMap();
+                dbg->Register(vchunk);
+                dumpAfter = vchunk->DebugPrintDirtyMap();
+                return true;
+            })
+            .GetValue(WaitTimeout);
+
+        // The Oracle is born at the connection count - the grown slot H5 is
+        // already present.
+        UNIT_ASSERT_STRING_CONTAINS(oracleDump, "H5");
+
+        // The grown host slot (H5) is absent in the vchunk before registering
+        // and present after - the DBG caught it up at registration.
+        UNIT_ASSERT_C(
+            dumpBefore.find("H5-") == TString::npos,
+            "unexpected H5 before register:\n" + dumpBefore);
+        UNIT_ASSERT_STRING_CONTAINS(dumpAfter, "H5-");
+    }
 }
 
 Y_UNIT_TEST_SUITE(TDDiskSessionSeqNoTest)
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.cpp
index 44f7fa95dba..99ac67e8c25 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.cpp
@@ -190,6 +190,10 @@ TDirectBlockGroupMock::TDirectBlockGroupMock()
         Y_ABORT_UNLESS(false, "Should set DumpHandler");
         return NThreading::TFuture<TDBGDumpResponse>();
     };
+    OnAddHostResultHandler = [](const auto&...)
+    {
+        Y_ABORT_UNLESS(false, "Should set OnAddHostResultHandler");
+    };
 }
 
 void TDirectBlockGroupMock::Register(TVChunkWeakPtr vChunk)
@@ -373,6 +377,19 @@ NThreading::TFuture<TDBGDumpResponse> TDirectBlockGroupMock::Dump()
     return DumpHandler();
 }
 
+void TDirectBlockGroupMock::OnAddHostResult(
+    const NProto::TError& error,
+    THostIndex newHostIndex,
+    NKikimrBlobStorage::NDDisk::TDDiskId ddiskId,
+    NKikimrBlobStorage::NDDisk::TDDiskId pbufferId)
+{
+    OnAddHostResultHandler(
+        error,
+        newHostIndex,
+        std::move(ddiskId),
+        std::move(pbufferId));
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.h
index ad233132a2a..440b024c91f 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/direct_block_group_mock.h
@@ -138,6 +138,12 @@ public:
     using TDBGDumpHandler =
         std::function<NThreading::TFuture<TDBGDumpResponse>()>;
 
+    using TOnAddHostResultHandler = std::function<void(
+        const NProto::TError& error,
+        THostIndex newHostIndex,
+        NKikimrBlobStorage::NDDisk::TDDiskId ddiskId,
+        NKikimrBlobStorage::NDDisk::TDDiskId pbufferId)>;
+
     TExecutorPtr Executor;
     TOracleMock Oracle;
     TScheduleHandler ScheduleHandler;
@@ -151,6 +157,7 @@ public:
     TDBGRestoreHandler RestoreDBGPBuffersHandler;
     TListPBuffersHandler ListPBuffersHandler;
     TDBGDumpHandler DumpHandler;
+    TOnAddHostResultHandler OnAddHostResultHandler;
 
     TVector<TVChunkWeakPtr> VChunks;
 
@@ -234,6 +241,12 @@ public:
         THostIndex hostIndex) override;
 
     NThreading::TFuture<TDBGDumpResponse> Dump() override;
+
+    void OnAddHostResult(
+        const NProto::TError& error,
+        THostIndex newHostIndex,
+        NKikimrBlobStorage::NDDisk::TDDiskId ddiskId,
+        NKikimrBlobStorage::NDDisk::TDDiskId pbufferId) override;
 };
 
 using TDirectBlockGroupMockPtr = std::shared_ptr<TDirectBlockGroupMock>;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.cpp
index 97b67d25d44..498bbd8b248 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.cpp
@@ -343,6 +343,16 @@ void TBlocksDirtyMap::UpdateConfig(const TVChunkConfig& vChunkConfig)
     }
 }
 
+void TBlocksDirtyMap::ResizeHosts(size_t newHostCount)
+{
+    Y_ABORT_UNLESS(newHostCount <= MaxHostCount);
+    Y_ABORT_UNLESS(newHostCount >= PBufferCounters.size());
+    Y_ABORT_UNLESS(newHostCount >= DDiskStates.size());
+
+    PBufferCounters.resize(newHostCount);
+    DDiskStates.resize(newHostCount);
+}
+
 TBlocksDirtyMap::~TBlocksDirtyMap()
 {
     Inflight.Enumerate(
@@ -716,6 +726,7 @@ std::optional<ui64> TBlocksDirtyMap::GetSafeBarrierForErase() const
 const TPBufferCounters& TBlocksDirtyMap::GetPBufferCounters(
     THostIndex host) const
 {
+    Y_ABORT_UNLESS(host < PBufferCounters.size());
     return PBufferCounters[host];
 }
 
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.h
index 352f548c9f1..b6edc999865 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map.h
@@ -232,6 +232,8 @@ public:
     // Note. Fresh watermarks are not applying for exists DDisks.
     void UpdateConfig(const TVChunkConfig& vChunkConfig);
 
+    void ResizeHosts(size_t newHostCount);
+
     void RestorePBuffer(ui64 lsn, TBlockRange64 range, THostIndex host);
 
     // MakeReadHint can work with multiple locations and returns multiple
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map_ut.cpp
index 90da23a359b..8c3d87abd33 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/dirty_map_ut.cpp
@@ -90,6 +90,27 @@ Y_UNIT_TEST_SUITE(TDirtyMapTest)
             readHint.DebugPrint());
     }
 
+    Y_UNIT_TEST(ShouldResizeHosts)
+    {
+        auto vchunkConfig = MakeTestVChunkConfig();
+        TBlocksDirtyMap dirtyMap(
+            vchunkConfig,
+            DefaultBlockSize,
+            DefaultVChunkSize / DefaultBlockSize);
+
+        vchunkConfig.AppendHost();
+        const auto newIdx = static_cast<THostIndex>(5);
+        dirtyMap.ResizeHosts(vchunkConfig.GetHostCount());
+        dirtyMap.UpdateConfig(vchunkConfig);
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            0u,
+            dirtyMap.GetPBufferCounters(newIdx).CurrentBytesCount);
+        UNIT_ASSERT_STRING_CONTAINS(
+            dirtyMap.DebugPrintDDiskState(),
+            "H5-{Disabled,0,0}");
+    }
+
     Y_UNIT_TEST(ShouldRespectWatermarksWhenConstruct)
     {
         auto vchunkConfig = MakeTestVChunkConfig();
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.cpp
index 204ef90f073..e18fa19a97d 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.cpp
@@ -375,6 +375,13 @@ void TFastPathService::UpdateVChunkConfig(const TVChunkConfig& cfg)
     ActorSystem->Send(PartitionActorId, event.release());
 }
 
+void TFastPathService::RequestAddHost(size_t directBlockGroupId)
+{
+    auto event = std::make_unique<TEvPartitionDirectPrivate::TEvAddHostToDBG>(
+        directBlockGroupId);
+    ActorSystem->Send(PartitionActorId, event.release());
+}
+
 ui64 TFastPathService::GenerateLsn()
 {
     const ui64 lsn = ++SequenceGenerator;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.h
index 26fcb37b7bc..256efd65050 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/fast_path_service.h
@@ -76,6 +76,12 @@ public:
     NThreading::TFuture<void> Run();
     NThreading::TFuture<void> Stop();
 
+    [[nodiscard]] const TVector<IDirectBlockGroupPtr>&
+    GetDirectBlockGroups() const
+    {
+        return DirectBlockGroups;
+    }
+
     // IStorage implementation
     NThreading::TFuture<TReadBlocksLocalResponse> ReadBlocksLocal(
         TCallContextPtr callContext,
@@ -102,6 +108,8 @@ public:
 
     void UpdateVChunkConfig(const TVChunkConfig& cfg) override;
 
+    void RequestAddHost(size_t directBlockGroupId) override;
+
     ui64 GenerateLsn() override;
 
     // Read-only info for the monitoring UI.
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.cpp
index 33f0af9d45e..dce48adff35 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.cpp
@@ -56,6 +56,13 @@ void THostRoles::SetRole(THostIndex host, EHostRole assignment)
     Assignments[host] = assignment;
 }
 
+void THostRoles::AppendRole(EHostRole assignment)
+{
+    Y_ABORT_UNLESS(Count < MaxHostCount);
+
+    Assignments[Count++] = assignment;
+}
+
 THostMask THostRoles::GetPrimary() const
 {
     THostMask result;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.h
index 3fa8d0ea59a..cb8b46835b2 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles.h
@@ -25,6 +25,8 @@ public:
     [[nodiscard]] EHostRole GetRole(THostIndex host) const;
     void SetRole(THostIndex host, EHostRole assignment);
 
+    void AppendRole(EHostRole assignment);
+
     [[nodiscard]] THostMask GetPrimary() const;
     [[nodiscard]] THostMask GetHandOff() const;
     [[nodiscard]] THostMask GetActive() const;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles_ut.cpp
index 4402bc76424..0e56a5eb38e 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_roles_ut.cpp
@@ -51,6 +51,20 @@ Y_UNIT_TEST_SUITE(THostRolesTest)
         UNIT_ASSERT(list.GetPrimary().Get(2));
         UNIT_ASSERT_VALUES_EQUAL(4u, list.GetActive().Count());
     }
+
+    Y_UNIT_TEST(ShouldAppendRole)
+    {
+        THostRoles roles(2);
+        roles.SetRole(0, EHostRole::Primary);
+        roles.SetRole(1, EHostRole::HandOff);
+
+        roles.AppendRole(EHostRole::None);
+
+        UNIT_ASSERT_VALUES_EQUAL(3u, roles.HostCount());
+        UNIT_ASSERT(roles.GetRole(0) == EHostRole::Primary);
+        UNIT_ASSERT(roles.GetRole(1) == EHostRole::HandOff);
+        UNIT_ASSERT(roles.GetRole(2) == EHostRole::None);
+    }
 }
 
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_state.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_state.h
index 6f05f101a45..7aef245fc5f 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_state.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_state.h
@@ -32,6 +32,8 @@ public:
 
     [[nodiscard]] virtual ui64 GetHostPBufferUsedSize(
         THostIndex hostIndex) const = 0;
+
+    virtual void QueryAddHost() = 0;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.cpp
index 81a2db7ca01..751f6690c64 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.cpp
@@ -208,6 +208,19 @@ void TOracle::Think(TInstant now)
     }
 }
 
+void TOracle::OnHostAdded()
+{
+    HostStatistics.emplace_back();
+    HostStates.emplace_back();
+    HostsHealths.push_back(EHostHealth::Online);
+    HostsReconnectDelays.emplace_back(MinReconnectDelay, MaxReconnectDelay);
+}
+
+size_t TOracle::GetHostCount() const
+{
+    return HostStates.size();
+}
+
 void TOracle::OnRequestStarted(
     THostIndex hostIndex,
     EOperation operation,
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.h
index 65fb5ab8a11..d26de62ba08 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle.h
@@ -121,6 +121,10 @@ public:
     [[nodiscard]] TDuration GetDDiskReconnectDelay(
         THostIndex hostIndex) override;
 
+    void OnHostAdded();
+
+    [[nodiscard]] size_t GetHostCount() const;
+
     [[nodiscard]] THostIndex SelectBestPBufferHost(
         THostMask hosts,
         EOperation operation) const override;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle_ut.cpp
index aef44ab809d..3708449e4d9 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/oracle_ut.cpp
@@ -39,6 +39,9 @@ struct THostStateControllerMock: public IHostStateController
 
         return HostPBufferUsedSize;
     }
+
+    void QueryAddHost() override
+    {}
 };
 
 TStorageConfigPtr MakeStorageConfig()
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.cpp
index 08ff001725c..01dcb1542b5 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.cpp
@@ -116,6 +116,18 @@ void TVChunkConfig::DisableHost(THostIndex hostIndex)
     EnabledHosts.Reset(hostIndex);
 }
 
+void TVChunkConfig::AppendHost()
+{
+    Y_ABORT_UNLESS(PBufferHosts.HostCount() == DDiskHosts.HostCount());
+    const auto newHostIndex = static_cast<THostIndex>(HostCount);
+
+    PBufferHosts.AppendRole(EHostRole::None);
+    DDiskHosts.AppendRole(EHostRole::None);
+    EnabledHosts.Reset(newHostIndex);
+    Watermarks.push_back(std::nullopt);
+    ++HostCount;
+}
+
 TString TVChunkConfig::EvacuateHost(THostIndex hostIndex)
 {
     DisableHost(hostIndex);
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.h
index 4069fee3da6..f44d114198d 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.h
@@ -35,6 +35,8 @@ public:
     // host.
     void DisableHost(THostIndex hostIndex);
 
+    void AppendHost();
+
     // Disables the host. Demote ddisk and pbuffer. If possible, adds ddisk on
     // the new host. Returns the text of the error or message to be logged.
     TString EvacuateHost(THostIndex hostIndex);
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config_ut.cpp
index 03e68a7f1bb..c8bf0c49a22 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config_ut.cpp
@@ -79,6 +79,22 @@ Y_UNIT_TEST_SUITE(TVChunkConfigTest)
         UNIT_ASSERT_VALUES_EQUAL("[H3,H4]", cfg.GetSecondaryPBuffers().Print());
         UNIT_ASSERT_VALUES_EQUAL("[H0,H1,H2]", cfg.GetHealthyDDisks().Print());
     }
+
+    Y_UNIT_TEST(ShouldAppendDisabledHost)
+    {
+        auto cfg = TVChunkConfig::MakeDefault(0, 3, 2);
+        const size_t before = cfg.GetHostCount();
+
+        cfg.AppendHost();
+
+        const auto newIdx = static_cast<THostIndex>(before);
+        UNIT_ASSERT_VALUES_EQUAL(before + 1, cfg.GetHostCount());
+        UNIT_ASSERT(cfg.GetPBufferRole(newIdx) == EHostRole::None);
+        UNIT_ASSERT(cfg.GetDDiskRole(newIdx) == EHostRole::None);
+        UNIT_ASSERT(cfg.GetDisabledHosts().Get(newIdx));
+        UNIT_ASSERT(!cfg.GetWatermark(newIdx).has_value());
+        UNIT_ASSERT(!cfg.GetDDisks().Get(newIdx));
+    }
 }
 
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_add_host_to_dbg.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_add_host_to_dbg.cpp
new file mode 100644
index 00000000000..ae89f223c21
--- /dev/null
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_add_host_to_dbg.cpp
@@ -0,0 +1,343 @@
+#include "fast_path_service.h"
+#include "part_database.h"
+#include "partition_direct_actor.h"
+#include "partition_direct_events_private.h"
+
+#include <ydb/core/nbs/cloud/storage/core/libs/common/error.h>
+
+#include <ydb/library/actors/core/log.h>
+#include <ydb/library/services/services.pb.h>
+
+namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect {
+
+using namespace NActors;
+using namespace NKikimr;
+using namespace NKikimr::NTabletFlatExecutor;
+
+using TDirectBlockGroupsConnections =
+    ::NYdb::NBS::PartitionDirect::NProto::TDirectBlockGroupsConnections;
+
+////////////////////////////////////////////////////////////////////////////////
+
+bool TPartitionActor::PrepareAddHostToDBG(
+    const TActorContext& ctx,
+    TTransactionContext& tx,
+    TTxPartition::TAddHostToDBG& args)
+{
+    Y_UNUSED(ctx);
+    Y_UNUSED(tx);
+    Y_UNUSED(args);
+
+    return true;
+}
+
+void TPartitionActor::ExecuteAddHostToDBG(
+    const TActorContext& ctx,
+    TTransactionContext& tx,
+    TTxPartition::TAddHostToDBG& args)
+{
+    Y_UNUSED(ctx);
+
+    TPartitionDatabase db(tx.DB);
+    // Store the connection and clear the intent in one tx, so they commit
+    // together: recovery never sees a half-applied add.
+    db.StoreDirectBlockGroupsConnections(args.DirectBlockGroupsConnections);
+    db.ClearAddHostInProgress();
+}
+
+void TPartitionActor::CompleteAddHostToDBG(
+    const TActorContext& ctx,
+    TTxPartition::TAddHostToDBG& args)
+{
+    const size_t dbgId = args.DirectBlockGroupId;
+
+    LOG_INFO(
+        ctx,
+        NKikimrServices::NBS_PARTITION,
+        "%s AddHost persisted dbgId=%lu newHostIndex=%s",
+        LogTitle.GetWithTime().c_str(),
+        dbgId,
+        PrintHostIndex(args.NewHostIndex).c_str());
+
+    Y_ABORT_UNLESS(FastPathService);
+
+    const auto& directBlockGroups = FastPathService->GetDirectBlockGroups();
+    Y_ABORT_UNLESS(dbgId < directBlockGroups.size());
+
+    // The new connection was persisted at NewHostIndex; read its DDisk/PBuffer
+    // back out to hand to the DBG.
+    const auto& newConnection =
+        args.DirectBlockGroupsConnections.GetDirectBlockGroupConnections(dbgId)
+            .GetConnections(args.NewHostIndex);
+
+    auto dbgPtr = directBlockGroups[dbgId];
+    auto executor = dbgPtr->GetExecutor();
+    executor->ExecuteSimple(
+        [dbgPtr,
+         newHostIndex = args.NewHostIndex,
+         newDDiskId = newConnection.GetDDiskId(),
+         newPBufferId = newConnection.GetPersistentBufferDDiskId()]() mutable
+        {
+            dbgPtr->OnAddHostResult(
+                {},
+                newHostIndex,
+                std::move(newDDiskId),
+                std::move(newPBufferId));
+        });
+
+    AddHostInFlight.reset();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+bool TPartitionActor::PrepareStartAddHost(
+    const TActorContext& ctx,
+    TTransactionContext& tx,
+    TTxPartition::TStartAddHost& args)
+{
+    Y_UNUSED(ctx);
+    Y_UNUSED(tx);
+    Y_UNUSED(args);
+
+    return true;
+}
+
+void TPartitionActor::ExecuteStartAddHost(
+    const TActorContext& ctx,
+    TTransactionContext& tx,
+    TTxPartition::TStartAddHost& args)
+{
+    Y_UNUSED(ctx);
+
+    TPartitionDatabase db(tx.DB);
+
+    TTxPartition::TAddHostInProgress proto;
+    proto.SetDirectBlockGroupId(static_cast<ui32>(args.DirectBlockGroupId));
+    proto.SetNewHostIndex(static_cast<ui32>(args.NewHostIndex));
+    db.StoreAddHostInProgress(proto);
+}
+
+void TPartitionActor::CompleteStartAddHost(
+    const TActorContext& ctx,
+    TTxPartition::TStartAddHost& args)
+{
+    SendAllocateDDiskForAddHost(
+        ctx,
+        args.DirectBlockGroupId,
+        args.NewHostIndex);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+namespace {
+
+// Validates a BSController add-host allocation response and, on success,
+// appends the newly granted DDisk/PBuffer to `result` (a copy of `current`).
+// Returns a retriable error - never aborts - for a failed or malformed
+// response: the add-host intent stays persisted and is replayed on recovery, so
+// a bad response must not crash-loop the tablet.
+NProto::TError AddConnection(
+    const TDirectBlockGroupsConnections& current,
+    const TEvBlobStorage::TEvControllerAllocateDDiskBlockGroupResult& msg,
+    size_t dbgId,
+    ui32 expectedCurrent,
+    TDirectBlockGroupsConnections* result)
+{
+    const auto& record = msg.Record;
+
+    if (record.GetStatus() != NKikimrProto::EReplyStatus::OK) {
+        return MakeError(
+            E_REJECTED,
+            TStringBuilder()
+                << "BSController error: " << record.GetErrorReason());
+    }
+
+    if (record.DirectBlockGroupsSize() != 1) {
+        return MakeError(
+            E_REJECTED,
+            TStringBuilder()
+                << "BSController returned " << record.DirectBlockGroupsSize()
+                << " DirectBlockGroups, expected 1");
+    }
+
+    const auto& group = record.GetDirectBlockGroups(0);
+    if (group.GetDirectBlockGroupId() != dbgId) {
+        return MakeError(
+            E_REJECTED,
+            "BSController response is for a different DirectBlockGroup");
+    }
+
+    if (group.GetError()) {
+        return MakeError(
+            E_REJECTED,
+            "BSController reported an error for this DirectBlockGroup");
+    }
+
+    if (static_cast<ui32>(group.DDiskIdSize()) != expectedCurrent + 1 ||
+        static_cast<ui32>(group.PersistentBufferDDiskIdSize()) !=
+            expectedCurrent + 1)
+    {
+        return MakeError(
+            E_REJECTED,
+            TStringBuilder()
+                << "BSController returned " << group.DDiskIdSize()
+                << " ddisks / " << group.PersistentBufferDDiskIdSize()
+                << " pbuffers, expected " << (expectedCurrent + 1));
+    }
+
+    const auto& newDDiskId = group.GetDDiskId(expectedCurrent);
+    const auto& newPBufferId =
+        group.GetPersistentBufferDDiskId(expectedCurrent);
+
+    const TString newDDiskIdBytes = newDDiskId.SerializeAsString();
+    const TString newPBufferIdBytes = newPBufferId.SerializeAsString();
+    for (const auto& conn:
+         current.GetDirectBlockGroupConnections(dbgId).GetConnections())
+    {
+        if (conn.GetDDiskId().SerializeAsString() == newDDiskIdBytes ||
+            conn.GetPersistentBufferDDiskId().SerializeAsString() ==
+                newPBufferIdBytes)
+        {
+            return MakeError(
+                E_REJECTED,
+                "BSController returned a DDisk/PBuffer already in this DBG");
+        }
+    }
+
+    *result = current;
+    auto* connection =
+        result->MutableDirectBlockGroupConnections(dbgId)->AddConnections();
+    connection->MutableDDiskId()->CopyFrom(newDDiskId);
+    connection->MutablePersistentBufferDDiskId()->CopyFrom(newPBufferId);
+    return {};
+}
+
+}   // namespace
+
+////////////////////////////////////////////////////////////////////////////////
+
+void TPartitionActor::HandleAddHostAllocationResult(
+    const TEvBlobStorage::TEvControllerAllocateDDiskBlockGroupResult::TPtr& ev,
+    const NActors::TActorContext& ctx)
+{
+    const auto* msg = ev->Get();
+    const size_t dbgId = ev->Cookie;
+
+    if (!AddHostInFlight.has_value() ||
+        AddHostInFlight->DirectBlockGroupId != dbgId)
+    {
+        LOG_WARN(
+            ctx,
+            NKikimrServices::NBS_PARTITION,
+            "%s AddHost response for unexpected dbgId=%lu (stale)",
+            LogTitle.GetWithTime().c_str(),
+            dbgId);
+        return;
+    }
+
+    const ui32 expectedCurrent = AddHostInFlight->NewHostIndex;
+    const auto newHostIndex = AddHostInFlight->NewHostIndex;
+    NTabletPipe::CloseClient(ctx, AddHostInFlight->BSPipeClient);
+
+    TDirectBlockGroupsConnections updated;
+    if (auto error = AddConnection(
+            DirectBlockGroupsConnections,
+            *msg,
+            dbgId,
+            expectedCurrent,
+            &updated);
+        HasError(error))
+    {
+        // Not cancelled: the intent stays persisted, so it is retried on the
+        // next recovery until BSController grants the DDisk.
+        LOG_WARN(
+            ctx,
+            NKikimrServices::NBS_PARTITION,
+            "%s AddHost (dbgId=%lu) not completed, kept for retry on "
+            "recovery: %s",
+            LogTitle.GetWithTime().c_str(),
+            dbgId,
+            FormatError(error).c_str());
+        return;
+    }
+
+    DirectBlockGroupsConnections = updated;
+
+    ExecuteTx(
+        ctx,
+        CreateTx<TAddHostToDBG>(std::move(updated), dbgId, newHostIndex));
+}
+
+void TPartitionActor::RejectAddHost(
+    const NActors::TActorContext& ctx,
+    size_t dbgId,
+    const TString& message)
+{
+    auto error = MakeError(E_REJECTED, message);
+
+    LOG_ERROR(
+        ctx,
+        NKikimrServices::NBS_PARTITION,
+        "%s AddHost failed (dbgId=%lu): %s",
+        LogTitle.GetWithTime().c_str(),
+        dbgId,
+        FormatError(error).c_str());
+
+    // Notify the DBG that asked for the host. dbgId is always a valid request
+    // index (see ValidateAddHostToDBGRequest), so the DBG exists.
+    const auto& directBlockGroups = FastPathService->GetDirectBlockGroups();
+    Y_ABORT_UNLESS(
+        dbgId < directBlockGroups.size(),
+        "RejectAddHost for a non-existent DBG (dbgId=%lu)",
+        dbgId);
+    auto dbgPtr = directBlockGroups[dbgId];
+    auto executor = dbgPtr->GetExecutor();
+    executor->ExecuteSimple([dbgPtr, error]()
+                            { dbgPtr->OnAddHostResult(error, 0, {}, {}); });
+}
+
+bool TPartitionActor::ValidateAddHostToDBGRequest(
+    const TActorContext& ctx,
+    size_t dbgId)
+{
+    // The request always carries a DBG's own index (QueryAddHost ->
+    // RequestAddHost), so an out-of-range dbgId is a bug, not a bad request.
+    const auto dbgCount = static_cast<size_t>(
+        DirectBlockGroupsConnections.DirectBlockGroupConnectionsSize());
+    Y_ABORT_UNLESS(
+        dbgId < dbgCount,
+        "AddHost for out-of-range dbgId=%lu (have %lu DBGs)",
+        dbgId,
+        dbgCount);
+
+    if (AddHostInFlight.has_value()) {
+        RejectAddHost(ctx, dbgId, "Another AddHost is already in progress");
+        return false;
+    }
+
+    // Authoritative AddHost gate: reads the persisted connection count under
+    // the single-in-flight guard above, so it cannot overshoot MaxHostCount or
+    // race a concurrent add. The DBG's own DDiskConnections lags, so it cannot
+    // gate.
+    const auto& dbgConn =
+        DirectBlockGroupsConnections.GetDirectBlockGroupConnections(dbgId);
+    const auto currentSize = static_cast<ui32>(dbgConn.GetConnections().size());
+
+    if (currentSize >= MaxHostCount) {
+        RejectAddHost(
+            ctx,
+            dbgId,
+            TStringBuilder() << "MaxHostCount=" << MaxHostCount << " reached");
+        return false;
+    }
+    if (currentSize == 0) {
+        RejectAddHost(ctx, dbgId, "AddHost on an empty DBG is not supported");
+        return false;
+    }
+
+    return true;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+}   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.cpp
index 7a4e7634d90..4482beb204b 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.cpp
@@ -179,4 +179,44 @@ void TPartitionDatabase::StoreVChunkConfig(const TVChunkConfig& cfg)
 
 ////////////////////////////////////////////////////////////////////////////////
 
+bool TPartitionDatabase::ReadAddHostInProgress(
+    TMaybe<TAddHostInProgress>& addHostInProgress)
+{
+    using TTable = TPartitionSchema::TabletInfo;
+
+    auto it = Table<TTable>().Key(1).Select<TTable::AddHostInProgress>();
+
+    if (!it.IsReady()) {
+        return false;
+    }
+
+    if (it.IsValid() && it.HaveValue<TTable::AddHostInProgress>()) {
+        addHostInProgress = it.GetValue<TTable::AddHostInProgress>();
+    }
+
+    return true;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+void TPartitionDatabase::StoreAddHostInProgress(
+    const TAddHostInProgress& addHostInProgress)
+{
+    using TTable = TPartitionSchema::TabletInfo;
+
+    Table<TTable>().Key(1).Update(
+        NKikimr::NIceDb::TUpdate<TTable::AddHostInProgress>(addHostInProgress));
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+void TPartitionDatabase::ClearAddHostInProgress()
+{
+    using TTable = TPartitionSchema::TabletInfo;
+
+    Table<TTable>().Key(1).UpdateToNull<TTable::AddHostInProgress>();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.h
index b057d85f476..f044a6ebe21 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database.h
@@ -14,6 +14,8 @@ class TPartitionDatabase: public NKikimr::NIceDb::TNiceDb
 {
     using TDirectBlockGroupsConnections =
         ::NYdb::NBS::PartitionDirect::NProto::TDirectBlockGroupsConnections;
+    using TAddHostInProgress =
+        ::NYdb::NBS::PartitionDirect::NProto::TAddHostInProgress;
 
 public:
     enum class EBlobIndexScanProgress
@@ -45,6 +47,10 @@ public:
         const TDirectBlockGroupsConnections& directBlockGroupsConnections);
 
     void StoreVChunkConfig(const TVChunkConfig& cfg);
+
+    bool ReadAddHostInProgress(TMaybe<TAddHostInProgress>& addHostInProgress);
+    void StoreAddHostInProgress(const TAddHostInProgress& addHostInProgress);
+    void ClearAddHostInProgress();
 };
 
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database_ut.cpp
index af250c67c3d..b71e280da1d 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_database_ut.cpp
@@ -12,6 +12,8 @@ namespace {
 
 using TDirectBlockGroupsConnections =
     ::NYdb::NBS::PartitionDirect::NProto::TDirectBlockGroupsConnections;
+using TAddHostInProgress =
+    ::NYdb::NBS::PartitionDirect::NProto::TAddHostInProgress;
 
 using NYdb::NBS::NBlockStore::NStorage::TTestExecutor;
 
@@ -147,6 +149,67 @@ Y_UNIT_TEST_SUITE(TPartitionDatabaseTest)
             });
     }
 
+    Y_UNIT_TEST(ShouldStoreReadAndClearAddHostInProgress)
+    {
+        TTestExecutor executor;
+
+        executor.WriteTx(
+            [&](NKikimr::NTable::TDatabase& db)
+            {
+                TPartitionDatabase partitionDb(db);
+                partitionDb.InitSchema();
+            });
+
+        // Absent right after init.
+        executor.ReadTx(
+            [&](NKikimr::NTable::TDatabase& db)
+            {
+                TPartitionDatabase partitionDb(db);
+                TMaybe<TAddHostInProgress> loaded;
+                UNIT_ASSERT(partitionDb.ReadAddHostInProgress(loaded));
+                UNIT_ASSERT(!loaded.Defined());
+            });
+
+        executor.WriteTx(
+            [&](NKikimr::NTable::TDatabase& db)
+            {
+                TPartitionDatabase partitionDb(db);
+                TAddHostInProgress intent;
+                intent.SetDirectBlockGroupId(3);
+                intent.SetNewHostIndex(5);
+                partitionDb.StoreAddHostInProgress(intent);
+            });
+
+        // Read back what was stored.
+        executor.ReadTx(
+            [&](NKikimr::NTable::TDatabase& db)
+            {
+                TPartitionDatabase partitionDb(db);
+                TMaybe<TAddHostInProgress> loaded;
+                UNIT_ASSERT(partitionDb.ReadAddHostInProgress(loaded));
+                UNIT_ASSERT(loaded.Defined());
+                UNIT_ASSERT_VALUES_EQUAL(3u, loaded->GetDirectBlockGroupId());
+                UNIT_ASSERT_VALUES_EQUAL(5u, loaded->GetNewHostIndex());
+            });
+
+        executor.WriteTx(
+            [&](NKikimr::NTable::TDatabase& db)
+            {
+                TPartitionDatabase partitionDb(db);
+                partitionDb.ClearAddHostInProgress();
+            });
+
+        // Absent again after clear.
+        executor.ReadTx(
+            [&](NKikimr::NTable::TDatabase& db)
+            {
+                TPartitionDatabase partitionDb(db);
+                TMaybe<TAddHostInProgress> loaded;
+                UNIT_ASSERT(partitionDb.ReadAddHostInProgress(loaded));
+                UNIT_ASSERT(!loaded.Defined());
+            });
+    }
+
     Y_UNIT_TEST(ShouldStoreAndReadVChunkConfigsPerRow)
     {
         TTestExecutor executor;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_loadstate.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_loadstate.cpp
index 340fdfa34f3..dd547eb5107 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_loadstate.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_loadstate.cpp
@@ -25,6 +25,7 @@ bool TPartitionActor::PrepareLoadState(
         db.ReadVolumeConfig(args.VolumeConfig),
         db.ReadDirectBlockGroupsConnections(args.DirectBlockGroupsConnections),
         db.ReadAllVChunkConfigs(args.VChunkConfigs),
+        db.ReadAddHostInProgress(args.AddHostInProgress),
     };
 
     bool ready = std::accumulate(
@@ -54,10 +55,23 @@ void TPartitionActor::CompleteLoadState(
         VolumeConfig = *args.VolumeConfig;
 
         if (args.DirectBlockGroupsConnections.Defined()) {
+            DDiskBlockGroupAllocated = true;
             Start(
                 ctx,
                 std::move(*args.DirectBlockGroupsConnections),
                 std::move(args.VChunkConfigs));
+
+            // An add-host was in flight at the last restart: hold the single
+            // in-flight slot and replay the BSController request once the fast
+            // path service is ready (see HandleFastPathServiceReady).
+            if (args.AddHostInProgress.Defined()) {
+                AddHostInFlight = TAddHostInFlight{
+                    .DirectBlockGroupId =
+                        args.AddHostInProgress->GetDirectBlockGroupId(),
+                    .NewHostIndex = static_cast<THostIndex>(
+                        args.AddHostInProgress->GetNewHostIndex()),
+                };
+            }
         }
     }
 }
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_schema.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_schema.h
index 3721ffa7f1c..f8b57ffd5a4 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_schema.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_schema.h
@@ -44,12 +44,22 @@ struct TPartitionSchema: public NKikimr::NIceDb::Schema
                 TDirectBlockGroupsConnections;
         };
 
+        // Set while an AddHost is in flight, cleared once it finishes (or is
+        // abandoned). Replayed on restart to recover a half-committed add.
+        struct AddHostInProgress
+            : public Column<5, NKikimr::NScheme::NTypeIds::String>
+        {
+            using Type =
+                ::NYdb::NBS::PartitionDirect::NProto::TAddHostInProgress;
+        };
+
         using TKey = TableKey<Id>;
         using TColumns = TableColumns<
             Id,
             StorageConfig,
             VolumeConfig,
-            DirectBlockGroupsConnections>;
+            DirectBlockGroupsConnections,
+            AddHostInProgress>;
 
         using StoragePolicy = TStoragePolicy<IndexChannel>;
     };
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_tx.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_tx.h
index 3014063e2e8..480b6c8fe60 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_tx.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/part_tx.h
@@ -1,10 +1,14 @@
 #pragma once
 
+#include <ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host.h>
 #include <ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/vchunk_config.h>
 #include <ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/protos/partition_direct.pb.h>
 
+#include <ydb/core/protos/blobstorage_ddisk.pb.h>
 #include <ydb/core/protos/blockstore_config.pb.h>
 
+#include <ydb/library/actors/core/actorid.h>
+
 #include <util/generic/maybe.h>
 #include <util/generic/vector.h>
 #include <util/system/types.h>
@@ -18,7 +22,9 @@ namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect {
     xxx(LoadState, __VA_ARGS__)                     \
     xxx(StoreVolumeConfig, __VA_ARGS__)             \
     xxx(StorePartitionIds, __VA_ARGS__)             \
-    xxx(UpdateVChunkConfig, __VA_ARGS__)
+    xxx(UpdateVChunkConfig, __VA_ARGS__)            \
+    xxx(StartAddHost, __VA_ARGS__)                  \
+    xxx(AddHostToDBG, __VA_ARGS__)
 
 // BLOCKSTORE_PARTITION_TRANSACTIONS
 
@@ -28,6 +34,8 @@ struct TTxPartition
 {
     using TDirectBlockGroupsConnections =
         ::NYdb::NBS::PartitionDirect::NProto::TDirectBlockGroupsConnections;
+    using TAddHostInProgress =
+        ::NYdb::NBS::PartitionDirect::NProto::TAddHostInProgress;
 
     //
     // InitSchema
@@ -51,6 +59,7 @@ struct TTxPartition
         TMaybe<NKikimrBlockStore::TVolumeConfig> VolumeConfig;
         TMaybe<TDirectBlockGroupsConnections> DirectBlockGroupsConnections;
         TVector<TVChunkConfig> VChunkConfigs;
+        TMaybe<TAddHostInProgress> AddHostInProgress;
 
         explicit TLoadState()
         {}
@@ -60,6 +69,7 @@ struct TTxPartition
             VolumeConfig.Clear();
             DirectBlockGroupsConnections.Clear();
             VChunkConfigs.clear();
+            AddHostInProgress.Clear();
         }
     };
 
@@ -117,6 +127,43 @@ struct TTxPartition
             // nothing to do
         }
     };
+
+    //
+    // TStartAddHost
+    //
+    struct TStartAddHost
+    {
+        const size_t DirectBlockGroupId;
+        const THostIndex NewHostIndex;
+
+        TStartAddHost(size_t directBlockGroupId, THostIndex newHostIndex)
+            : DirectBlockGroupId(directBlockGroupId)
+            , NewHostIndex(newHostIndex)
+        {}
+
+        void Clear()
+        {}
+    };
+
+    struct TAddHostToDBG
+    {
+        const TDirectBlockGroupsConnections DirectBlockGroupsConnections;
+        const size_t DirectBlockGroupId;
+        const THostIndex NewHostIndex;
+
+        TAddHostToDBG(
+            TDirectBlockGroupsConnections directBlockGroupsConnections,
+            size_t directBlockGroupId,
+            THostIndex newHostIndex)
+            : DirectBlockGroupsConnections(
+                  std::move(directBlockGroupsConnections))
+            , DirectBlockGroupId(directBlockGroupId)
+            , NewHostIndex(newHostIndex)
+        {}
+
+        void Clear()
+        {}
+    };
 };
 
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.cpp
index 3205055432f..c62fd784993 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.cpp
@@ -270,6 +270,8 @@ void TPartitionActor::Start(
         vChunkConfigsByIndex[cfg.GetVChunkIndex()] = cfg;
     }
 
+    DirectBlockGroupsConnections = directBlockGroupsConnections;
+
     const ui64 blockCount = VolumeConfig.GetPartitions(0).GetBlockCount();
     FastPathService = std::make_shared<TFastPathService>(
         TActivationContext::ActorSystem(),
@@ -312,6 +314,22 @@ void TPartitionActor::HandleFastPathServiceReady(
         "%s All DBGs reached initial locked quorum, opening endpoint",
         LogTitle.GetWithTime().c_str());
 
+    // Re-send the BSC request for an add-host in flight at the last restart
+    // (no live add can be in flight this early). BSController is idempotent.
+    if (AddHostInFlight.has_value()) {
+        LOG_INFO(
+            ctx,
+            NKikimrServices::NBS_PARTITION,
+            "%s Replaying in-flight AddHost dbgId=%lu newHostIndex=%s",
+            LogTitle.GetWithTime().c_str(),
+            AddHostInFlight->DirectBlockGroupId,
+            PrintHostIndex(AddHostInFlight->NewHostIndex).c_str());
+        SendAllocateDDiskForAddHost(
+            ctx,
+            AddHostInFlight->DirectBlockGroupId,
+            AddHostInFlight->NewHostIndex);
+    }
+
     LoadActorAdapter = CreateLoadActorAdapter(ctx.SelfID, FastPathService);
 
     {
@@ -417,14 +435,27 @@ void TPartitionActor::HandleControllerAllocateDDiskBlockGroupResult(
     const TEvBlobStorage::TEvControllerAllocateDDiskBlockGroupResult::TPtr& ev,
     const NActors::TActorContext& ctx)
 {
-    const auto* msg = ev->Get();
-
     LOG_INFO(
         ctx,
         NKikimrServices::NBS_PARTITION,
         "%s HandleControllerAllocateDDiskBlockGroupResult record is: %s",
         LogTitle.GetWithTime().c_str(),
-        msg->Record.DebugString().data());
+        ev->Get()->Record.DebugString().data());
+
+    // The first allocation response sets up the group; any later one is the
+    // result of an add-host request.
+    if (DDiskBlockGroupAllocated) {
+        HandleAddHostAllocationResult(ev, ctx);
+    } else {
+        HandleInitialAllocationResult(ev, ctx);
+    }
+}
+
+void TPartitionActor::HandleInitialAllocationResult(
+    const TEvBlobStorage::TEvControllerAllocateDDiskBlockGroupResult::TPtr& ev,
+    const NActors::TActorContext& ctx)
+{
+    const auto* msg = ev->Get();
 
     if (msg->Record.GetStatus() == NKikimrProto::EReplyStatus::OK) {
         Y_ABORT_UNLESS(
@@ -444,7 +475,7 @@ void TPartitionActor::HandleControllerAllocateDDiskBlockGroupResult(
             }
         }
 
-        DdiskBlockGroupAllocated = true;
+        DDiskBlockGroupAllocated = true;
         ExecuteTx(ctx, CreateTx<TStorePartitionIds>(std::move(ids)));
     } else {
         LOG_ERROR(
@@ -460,6 +491,83 @@ void TPartitionActor::HandleControllerAllocateDDiskBlockGroupResult(
     NTabletPipe::CloseClient(ctx, BSControllerPipeClient);
 }
 
+void TPartitionActor::HandleAddHostToDBG(
+    const TEvPartitionDirectPrivate::TEvAddHostToDBG::TPtr& ev,
+    const NActors::TActorContext& ctx)
+{
+    const auto* msg = ev->Get();
+    const auto dbgId = msg->DirectBlockGroupId;
+
+    LOG_INFO(
+        ctx,
+        NKikimrServices::NBS_PARTITION,
+        "%s Handle AddHostToDBG dbgId=%lu",
+        LogTitle.GetWithTime().c_str(),
+        dbgId);
+
+    // TEvAddHostToDBG is only sent by a running FastPathService, so it (and the
+    // allocated DBGs) is alive by the time we handle the request.
+    Y_ABORT_UNLESS(FastPathService);
+
+    if (!ValidateAddHostToDBGRequest(ctx, dbgId)) {
+        return;
+    }
+
+    const auto& dbgConn =
+        DirectBlockGroupsConnections.GetDirectBlockGroupConnections(dbgId);
+    const auto currentSize = static_cast<ui32>(dbgConn.GetConnections().size());
+
+    // Persist the intent before the BSController request (sent from the tx's
+    // completion). A crash after the DDisk is allocated but before the
+    // connection is persisted then leaves a durable intent, replayed on
+    // restart.
+    AddHostInFlight = TAddHostInFlight{
+        .DirectBlockGroupId = dbgId,
+        .NewHostIndex = static_cast<THostIndex>(currentSize),
+    };
+
+    ExecuteTx(
+        ctx,
+        CreateTx<TStartAddHost>(dbgId, static_cast<THostIndex>(currentSize)));
+}
+
+void TPartitionActor::SendAllocateDDiskForAddHost(
+    const TActorContext& ctx,
+    size_t dbgId,
+    THostIndex newHostIndex)
+{
+    Y_ABORT_UNLESS(AddHostInFlight.has_value());
+
+    const ui64 blockCount = VolumeConfig.GetPartitions(0).GetBlockCount();
+    const ui64 regionsCount =
+        AlignUp(blockCount * VolumeConfig.GetBlockSize(), RegionSize) /
+        RegionSize;
+
+    const auto pipe = ctx.Register(
+        NTabletPipe::CreateClient(ctx.SelfID, MakeBSControllerID()));
+    AddHostInFlight->BSPipeClient = pipe;
+
+    // Idempotent: NumDDisks=N+1 is the desired final state, not "add one"; a
+    // re-sent request returns the same DDisk from BSController's persisted
+    // allocation, so a retry (e.g. after a restart) is safe.
+    const ui32 numDDisks = static_cast<ui32>(newHostIndex) + 1;
+    auto request = std::make_unique<
+        TEvBlobStorage::TEvControllerAllocateDDiskBlockGroup>();
+    request->Record.SetDDiskPoolName(StorageConfig->GetDDiskPoolName());
+    request->Record.SetPersistentBufferDDiskPoolName(
+        StorageConfig->GetPersistentBufferDDiskPoolName());
+    request->Record.SetTabletId(TabletID());
+
+    auto* op = request->Record.AddDirectBlockGroupOperations();
+    op->SetDirectBlockGroupId(dbgId);
+    auto* define = op->MutableDefineDirectBlockGroup();
+    define->SetNumDDisks(numDDisks);
+    define->SetNumChunksPerDDisk(regionsCount);
+    define->SetNumPersistentBuffers(numDDisks);
+
+    NTabletPipe::SendData(ctx, pipe, request.release(), dbgId);
+}
+
 void TPartitionActor::HandleGetLoadActorAdapterActorId(
     const TEvService::TEvGetLoadActorAdapterActorIdRequest::TPtr& ev,
     const NActors::TActorContext& ctx)
@@ -485,7 +593,7 @@ void TPartitionActor::HandleUpdateVolumeConfig(
         LogTitle.GetWithTime().c_str(),
         msg->Record.GetVolumeConfig().GetVersion());
 
-    if (DdiskBlockGroupAllocated) {
+    if (DDiskBlockGroupAllocated) {
         LOG_ERROR(
             ctx,
             NKikimrServices::NBS_PARTITION,
@@ -572,6 +680,7 @@ STFUNC(TPartitionActor::StateWork)
         HFunc(
             TEvPartitionDirectPrivate::TEvFastPathServiceReady,
             HandleFastPathServiceReady);
+        HFunc(TEvPartitionDirectPrivate::TEvAddHostToDBG, HandleAddHostToDBG);
 
         HFunc(
             TEvPartitionDirectPrivate::TEvFastPathServiceShutdown,
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.h
index fe101579e41..47199aaacfb 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_actor.h
@@ -8,6 +8,7 @@
 #include <ydb/core/nbs/cloud/blockstore/libs/storage/api/service.h>
 #include <ydb/core/nbs/cloud/blockstore/libs/storage/core/tablet.h>
 #include <ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.h>
+#include <ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host.h>
 
 #include <ydb/core/nbs/cloud/storage/core/libs/common/error.h>
 #include <ydb/core/nbs/cloud/storage/core/libs/coroutine/executor_pool.h>
@@ -22,6 +23,8 @@
 #include <ydb/library/actors/core/mon.h>
 #include <ydb/library/services/services.pb.h>
 
+#include <optional>
+
 namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect {
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -49,9 +52,21 @@ private:
     NActors::TActorId BSControllerPipeClient;
 
     NActors::TActorId LoadActorAdapter;
-    bool DdiskBlockGroupAllocated = false;
+    bool DDiskBlockGroupAllocated = false;
     std::shared_ptr<TFastPathService> FastPathService;
 
+    TDirectBlockGroupsConnections DirectBlockGroupsConnections;
+
+    struct TAddHostInFlight
+    {
+        size_t DirectBlockGroupId = 0;
+        THostIndex NewHostIndex = InvalidHostIndex;
+        NActors::TActorId BSPipeClient;
+    };
+
+    // At most one add-host runs at a time across the whole partition.
+    std::optional<TAddHostInFlight> AddHostInFlight;
+
 public:
     TPartitionActor(
         const NActors::TActorId& tablet,
@@ -96,11 +111,29 @@ private:
 
     void AllocateDDiskBlockGroup(const NActors::TActorContext& ctx);
 
+    void SendAllocateDDiskForAddHost(
+        const NActors::TActorContext& ctx,
+        size_t dbgId,
+        THostIndex newHostIndex);
+
     void HandleControllerAllocateDDiskBlockGroupResult(
         const NKikimr::TEvBlobStorage::
             TEvControllerAllocateDDiskBlockGroupResult::TPtr& ev,
         const NActors::TActorContext& ctx);
 
+    // Sets up the group from the first (bulk) allocation response.
+    void HandleInitialAllocationResult(
+        const NKikimr::TEvBlobStorage::
+            TEvControllerAllocateDDiskBlockGroupResult::TPtr& ev,
+        const NActors::TActorContext& ctx);
+
+    // Applies a single add-host allocation response: validate, append the new
+    // connection, and persist it via TAddHostToDBG.
+    void HandleAddHostAllocationResult(
+        const NKikimr::TEvBlobStorage::
+            TEvControllerAllocateDDiskBlockGroupResult::TPtr& ev,
+        const NActors::TActorContext& ctx);
+
     void HandleGetLoadActorAdapterActorId(
         const NYdb::NBS::NBlockStore::TEvService::
             TEvGetLoadActorAdapterActorIdRequest::TPtr& ev,
@@ -126,6 +159,21 @@ private:
         const TEvPartitionDirectPrivate::TEvFastPathServiceStopped::TPtr& ev,
         const NActors::TActorContext& ctx);
 
+    void HandleAddHostToDBG(
+        const TEvPartitionDirectPrivate::TEvAddHostToDBG::TPtr& ev,
+        const NActors::TActorContext& ctx);
+
+    // Rejects (logs + notifies the DBG) and returns false if the AddHost
+    // request is invalid; true if it may proceed.
+    bool ValidateAddHostToDBGRequest(
+        const NActors::TActorContext& ctx,
+        size_t dbgId);
+
+    void RejectAddHost(
+        const NActors::TActorContext& ctx,
+        size_t dbgId,
+        const TString& message);
+
     void Start(
         const NActors::TActorContext& ctx,
         TDirectBlockGroupsConnections directBlockGroupsConnections,
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_events_private.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_events_private.h
index 4d0e4be2d9f..e79a7922695 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_events_private.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_events_private.h
@@ -33,6 +33,7 @@ struct TEvPartitionDirectPrivate
 
         EvFastPathServiceShutdown,
         EvFastPathServiceStopped,
+        EvAddHostToDBG,
 
         EvEnd,
     };
@@ -68,6 +69,16 @@ struct TEvPartitionDirectPrivate
               TEventLocal<TEvFastPathServiceStopped, EvFastPathServiceStopped>
     {
     };
+
+    struct TEvAddHostToDBG
+        : public NActors::TEventLocal<TEvAddHostToDBG, EvAddHostToDBG>
+    {
+        size_t DirectBlockGroupId;
+
+        explicit TEvAddHostToDBG(size_t dbgId)
+            : DirectBlockGroupId(dbgId)
+        {}
+    };
 };
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_ut.cpp
index 3d1ed757428..4ed6d4b3f85 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/partition_direct_ut.cpp
@@ -719,6 +719,145 @@ Y_UNIT_TEST_SUITE(TPartitionDirectTest)
         StopFastPathService(env, partition, edge);
     }
 
+    Y_UNIT_TEST(ShouldRequestDDiskAllocationForAddedHost)
+    {
+        TEnvironmentSetup env{{
+            .NodeCount = 8,
+            .Erasure = TBlobStorageGroupType::Erasure4Plus2Block,
+        }};
+        auto& runtime = env.Runtime;
+        runtime->SetLogPriority(
+            NKikimrServices::NBS_PARTITION,
+            NActors::NLog::PRI_DEBUG);
+
+        auto scopedService = SetupStorage(env, EWriteMode::DirectWrite);
+
+        // The add-host allocation is the only request that uses
+        // DirectBlockGroupOperations (the initial bulk allocation uses
+        // Queries).
+        ui32 addHostRequestCount = 0;
+        ui32 addHostNumDDisks = 0;
+        runtime->FilterFunction = [&](ui32, std::unique_ptr<IEventHandle>& ev)
+        {
+            if (ev->GetTypeRewrite() ==
+                TEvBlobStorage::TEvControllerAllocateDDiskBlockGroup::EventType)
+            {
+                auto* msg = ev->Get<
+                    TEvBlobStorage::TEvControllerAllocateDDiskBlockGroup>();
+                if (msg->Record.DirectBlockGroupOperationsSize() > 0) {
+                    ++addHostRequestCount;
+                    addHostNumDDisks =
+                        msg->Record.GetDirectBlockGroupOperations(0)
+                            .GetDefineDirectBlockGroup()
+                            .GetNumDDisks();
+                }
+            }
+            return true;
+        };
+
+        const ui64 partition = CreatePartitionTablet(env);
+
+        // Drop the throwaway sender right after sending: its strict edge actor
+        // would otherwise abort on the pipe notifications during the
+        // free-running Sim. The payload is still delivered by the pipe client.
+        const TActorId sender = runtime->AllocateEdgeActor(
+            env.Settings.ControllerNodeId,
+            __FILE__,
+            __LINE__);
+        env.Runtime->SendToPipe(
+            partition,
+            sender,
+            new TEvPartitionDirectPrivate::TEvAddHostToDBG(0),
+            0,
+            TTestActorSystem::GetPipeConfigWithRetries());
+        runtime->DestroyActor(sender);
+
+        env.Sim(TDuration::Seconds(10));
+
+        // The add persisted its intent and asked BSController to grow the group
+        // to DirectBlockGroupHostCount + 1 DDisks.
+        UNIT_ASSERT_VALUES_EQUAL(1u, addHostRequestCount);
+        UNIT_ASSERT_VALUES_EQUAL(
+            static_cast<ui32>(DirectBlockGroupHostCount + 1),
+            addHostNumDDisks);
+    }
+
+    Y_UNIT_TEST(ShouldReplayInFlightAddHostAfterRestart)
+    {
+        TEnvironmentSetup env{{
+            .NodeCount = 8,
+            .Erasure = TBlobStorageGroupType::Erasure4Plus2Block,
+        }};
+        auto& runtime = env.Runtime;
+        runtime->SetLogPriority(
+            NKikimrServices::NBS_PARTITION,
+            NActors::NLog::PRI_DEBUG);
+
+        auto scopedService = SetupStorage(env, EWriteMode::DirectWrite);
+
+        ui32 addHostRequestCount = 0;
+        bool dropNextAddHostResult = false;
+        runtime->FilterFunction = [&](ui32, std::unique_ptr<IEventHandle>& ev)
+        {
+            const auto type = ev->GetTypeRewrite();
+            if (type ==
+                TEvBlobStorage::TEvControllerAllocateDDiskBlockGroup::EventType)
+            {
+                auto* msg = ev->Get<
+                    TEvBlobStorage::TEvControllerAllocateDDiskBlockGroup>();
+                if (msg->Record.DirectBlockGroupOperationsSize() > 0) {
+                    ++addHostRequestCount;
+                }
+            }
+            // Drop the first add-host result so the connection is never
+            // persisted (the intent stays), forcing a replay on restart.
+            if (dropNextAddHostResult &&
+                type ==
+                    TEvBlobStorage::TEvControllerAllocateDDiskBlockGroupResult::
+                        EventType)
+            {
+                dropNextAddHostResult = false;
+                return false;
+            }
+            return true;
+        };
+
+        const ui64 partition = CreatePartitionTablet(env);
+
+        // Trigger an add whose BSController result is dropped: the intent is
+        // persisted but the connection is not.
+        dropNextAddHostResult = true;
+        {
+            const TActorId sender = runtime->AllocateEdgeActor(
+                env.Settings.ControllerNodeId,
+                __FILE__,
+                __LINE__);
+            env.Runtime->SendToPipe(
+                partition,
+                sender,
+                new TEvPartitionDirectPrivate::TEvAddHostToDBG(0),
+                0,
+                TTestActorSystem::GetPipeConfigWithRetries());
+            runtime->DestroyActor(sender);
+        }
+        env.Sim(TDuration::Seconds(10));
+        UNIT_ASSERT_VALUES_EQUAL(1u, addHostRequestCount);
+
+        // Restart: the persisted intent must be replayed.
+        {
+            scopedService.reset();
+            env.RestartNode(env.Settings.ControllerNodeId);
+            env.Sim(TDuration::Seconds(1));
+            scopedService = std::make_unique<TScopedNbsService>(
+                CreateNbsConfig(EWriteMode::DirectWrite));
+        }
+        WaitForTabletBoot(env);
+        env.Sim(TDuration::Seconds(10));
+
+        // The replay re-sent the BSController allocation request.
+        UNIT_ASSERT_VALUES_EQUAL(2u, addHostRequestCount);
+    }
+
     Y_UNIT_TEST(BasicWriteReadPBufferReplication)
     {
         BasicWriteRead(EWriteMode::IndirectWrite);
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/protos/partition_direct.proto b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/protos/partition_direct.proto
index d5105778fd1..0dcbd989dea 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/protos/partition_direct.proto
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/protos/partition_direct.proto
@@ -42,3 +42,9 @@ message TTabletInfo {
     NKikimrBlockStore.TVolumeConfig VolumeConfig = 2;
     TDirectBlockGroupsConnections DirectBlockGroupsConnections = 3;
 }
+
+// An in-flight AddHost, persisted so it can be finished (replayed) on restart.
+message TAddHostInProgress {
+    uint32 DirectBlockGroupId = 1;
+    uint32 NewHostIndex = 2;
+}
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.cpp
index 6622d221149..35db2484706 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.cpp
@@ -224,6 +224,33 @@ void TVChunk::SetHostState(THostIndex hostIndex, EHostState state)
     UpdateConfig(std::move(prepare), std::move(apply));
 }
 
+void TVChunk::OnHostAppended(size_t newHostCount)
+{
+    Y_ABORT_UNLESS(ExecutorThreadChecker.Check());
+
+    // Resize synchronously - the config apply below is async, but the new host
+    // is connected and used before it runs.
+    BlocksDirtyMap.ResizeHosts(newHostCount);
+
+    auto prepare = [weakSelf = weak_from_this()]() -> TVChunkConfig
+    {
+        if (auto self = weakSelf.lock()) {
+            TVChunkConfig cfg = self->VChunkConfig;
+            cfg.AppendHost();
+            return cfg;
+        }
+        return TVChunkConfig{};
+    };
+    auto apply = [weakSelf = weak_from_this()]()
+    {
+        if (auto self = weakSelf.lock()) {
+            self->ApplyConfig();
+        }
+    };
+
+    UpdateConfig(std::move(prepare), std::move(apply));
+}
+
 const TVChunkConfig& TVChunk::GetConfig() const
 {
     Y_ABORT_UNLESS(ExecutorThreadChecker.Check());
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.h
index 306ff9f8890..6760e1d2c74 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk.h
@@ -59,6 +59,8 @@ public:
 
     void SetHostState(THostIndex hostIndex, EHostState state);
 
+    void OnHostAppended(size_t newHostCount);
+
     [[nodiscard]] const TVChunkConfig& GetConfig() const;
     [[nodiscard]] ui64 GetPBufferUsedSize(THostIndex hostIndex) const;
     [[nodiscard]] TString DebugPrintDirtyMap();
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk_ut.cpp
index 9e8040ec2b9..bc1311fff0c 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/vchunk_ut.cpp
@@ -302,6 +302,71 @@ Y_UNIT_TEST_SUITE(TVChunkTest)
         onStop.GetValue(TDuration::Seconds(10));
     }
 
+    Y_UNIT_TEST_F(ShouldAppendHostAndGrowDirtyMap, TBaseFixture)
+    {
+        Init();
+
+        auto vchunk = std::make_shared<TVChunk>(
+            Runtime->GetActorSystem(0),
+            PartitionDirectService.get(),
+            VChunkConfig,
+            DirectBlockGroup,
+            3,
+            DefaultVChunkSize,
+            Counters);
+        vchunk->Start();
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            DirectBlockGroupHostCount,
+            AccessConfig(*vchunk).GetHostCount());
+
+        {
+            TPromise<void> ready = NewPromise();
+            auto wait = ready.GetFuture();
+            DirectBlockGroup->GetExecutor()->ExecuteSimple(
+                [&]()
+                {
+                    vchunk->OnHostAppended(DirectBlockGroupHostCount + 1);
+                    ready.SetValue();
+                });
+            wait.GetValue(TDuration::Seconds(10));
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            DirectBlockGroupHostCount,
+            AccessConfig(*vchunk).GetHostCount());
+
+        UNIT_ASSERT_STRING_CONTAINS(
+            AccessBlocksDirtyMap(*vchunk).DebugPrintDDiskState(),
+            "H5-{Disabled,0,0}");
+
+        {
+            UNIT_ASSERT_VALUES_EQUAL(1, ScheduledTasks.size());
+            auto task = RunScheduledTasks();
+            task.Wait(TDuration::Seconds(10));
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            DirectBlockGroupHostCount + 1,
+            AccessConfig(*vchunk).GetHostCount());
+        UNIT_ASSERT_STRING_CONTAINS(
+            AccessConfig(*vchunk).DebugPrint(),
+            "PBuffer{Primary;Primary;Primary;HandOff;HandOff;None}");
+        UNIT_ASSERT_STRING_CONTAINS(
+            AccessConfig(*vchunk).DebugPrint(),
+            "DDisk{Primary;Primary;Primary;None;None;None}");
+        UNIT_ASSERT_STRING_CONTAINS(
+            AccessConfig(*vchunk).DebugPrint(),
+            "Enabled{+++++-}");
+
+        UNIT_ASSERT_STRING_CONTAINS(
+            AccessBlocksDirtyMap(*vchunk).DebugPrintDDiskState(),
+            "H5-{Disabled,0,0}");
+
+        auto onStop = vchunk->Stop();
+        onStop.GetValue(TDuration::Seconds(10));
+    }
+
     Y_UNIT_TEST_F(ShouldSwitchHostToOfflineAndBack, TBaseFixture)
     {
         Init();
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/ya.make b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/ya.make
index 0b54a84130b..0d0652e6368 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/ya.make
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/ya.make
@@ -10,6 +10,7 @@ SRCS(
     fast_path_service.cpp
     flush_request.cpp
     load_actor_adapter.cpp
+    part_add_host_to_dbg.cpp
     part_database.cpp
     part_initschema.cpp
     part_loadstate.cpp