[NBS] Cool down flush requests when node disconnected (#45894)

31 files changed, 514 insertions, 172 deletions

diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.cpp
index b12d55d125f..aff8a007246 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.cpp
@@ -75,9 +75,8 @@ TString ToString(const TLogTitle::TVolume& data)
 {
     TStringBuilder stream;
 
-    stream << "[v:" << data.TabletId;
-    stream << " g:" << TOptional{data.Generation};
-    stream << " d:" << TOptional{data.DiskId};
+    stream << "[v:" << TOptional{data.DiskId} << "/" << data.TabletId << "/"
+           << data.Generation;
 
     return stream;
 }
@@ -86,9 +85,8 @@ TString ToString(const TLogTitle::TPartitionDirect& data)
 {
     TStringBuilder stream;
 
-    stream << "[pd:" << data.TabletId;
-    stream << " g:" << TOptional{data.Generation};
-    stream << " d:" << TOptional{data.DiskId};
+    stream << "[pd:" << TOptional{data.DiskId} << "/" << data.TabletId << "/"
+           << data.Generation;
 
     return stream;
 }
@@ -97,10 +95,8 @@ TString ToString(const TLogTitle::TDirectBlockGroup& data)
 {
     TStringBuilder stream;
 
-    stream << "[dbg:" << data.TabletId;
-    stream << " g:" << TOptional{data.Generation};
-    stream << " d:" << TOptional{data.DiskId};
-    stream << " indx:" << data.DirectBlockGroupIndex;
+    stream << "[dbg:" << data.DiskId << "/" << data.DBGIndex;
+    stream << " t:" << TOptional{data.TabletId} << "/" << data.Generation;
 
     return stream;
 }
@@ -109,8 +105,8 @@ TString ToString(const TLogTitle::TVChunk& data)
 {
     TStringBuilder stream;
 
-    stream << "[vchk:" << data.DiskId;
-    stream << " indx:" << data.VChunkIndex;
+    stream << "[vchk:" << data.DiskId << "/" << data.DBGIndex << "/"
+           << data.VChunkIndex;
 
     return stream;
 }
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.h b/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.h
index b7021add347..14f10657f41 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/model/log_title.h
@@ -22,7 +22,7 @@ public:
 
     struct TVolume
     {
-        ui64 TabletId;
+        ui64 TabletId = 0;
         TString DiskId;
         ui32 Generation = 0;
     };
@@ -37,15 +37,16 @@ public:
     struct TDirectBlockGroup
     {
         TString DiskId;
+        size_t DBGIndex = 0;
         ui64 TabletId = 0;
         ui32 Generation = 0;
-        size_t DirectBlockGroupIndex = 0;
     };
 
     struct TVChunk
     {
         TString DiskId;
-        ui32 VChunkIndex;
+        ui32 DBGIndex = 0;
+        ui32 VChunkIndex = 0;
     };
 
     struct TDDiskDataCopier
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 68fd981751e..66fea1c07c4 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
@@ -109,10 +109,9 @@ TDirectBlockGroup::TDirectBlockGroup(
           GetCycleCount(),
           TLogTitle::TDirectBlockGroup{
               .DiskId = diskId,
+              .DBGIndex = DirectBlockGroupIndex,
               .TabletId = TabletId,
-              .Generation = TabletGeneration,
-              .DirectBlockGroupIndex = DirectBlockGroupIndex,
-          })
+              .Generation = TabletGeneration})
     , Oracle(StorageConfig, this)
 {
     Y_ASSERT(pbufferIds.size() == DirectBlockGroupHostCount);
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 bd9fc26b7ef..44f7fa95dba 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
@@ -98,6 +98,12 @@ TDuration TOracleMock::GetIndirectWriteReplyTimeout() const
     return PBufferReplyTimeout;
 }
 
+TDuration TOracleMock::GetFlushRequestCooldown(THostMask hosts) const
+{
+    Y_UNUSED(hosts);
+    return FlushRequestCooldown;
+}
+
 TDuration TOracleMock::GetFlushRequestTimeout() const
 {
     return FlushRequestTimeout;
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 2028d9cde91..ad233132a2a 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
@@ -16,6 +16,7 @@ struct TOracleMock: public IOracle
     TDuration WriteRequestTimeout;
     TDuration PBufferReplyTimeout;
     EWriteMode WriteMode = EWriteMode::DirectWrite;
+    TDuration FlushRequestCooldown;
     TDuration FlushRequestTimeout;
     TDuration EraseRequestTimeout;
     TVector<THostStat> HostStatistics;
@@ -50,14 +51,19 @@ struct TOracleMock: public IOracle
         THostIndex host,
         EDataLocation dataLocation) const override;
     [[nodiscard]] TDuration GetReadRequestTimeout() const override;
+
+    [[nodiscard]] EWriteMode GetWriteMode() const override;
     [[nodiscard]] TDuration GetWriteHedgingDelay(
         THostMask hosts,
         bool indirect) const override;
     [[nodiscard]] TDuration GetWriteRequestTimeout() const override;
     [[nodiscard]] TDuration GetIndirectWriteReplyTimeout() const override;
+
+    [[nodiscard]] TDuration GetFlushRequestCooldown(
+        THostMask hosts) const override;
     [[nodiscard]] TDuration GetFlushRequestTimeout() const override;
+
     [[nodiscard]] TDuration GetEraseRequestTimeout() const override;
-    [[nodiscard]] EWriteMode GetWriteMode() const override;
 
     [[nodiscard]] const THostStat& GetHostStatistics(
         THostIndex hostIndex) const override;
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 be8c308cf9a..97b67d25d44 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
@@ -338,6 +338,8 @@ void TBlocksDirtyMap::UpdateConfig(const TVChunkConfig& vChunkConfig)
 
     for (auto lsn: erased) {
         Inflight.RemoveRange(lsn);
+        ReadyToErase.erase(lsn);
+        ReadyToFlush.erase(lsn);
     }
 }
 
@@ -572,12 +574,16 @@ void TBlocksDirtyMap::FlushFinished(
     const TVector<ui64>& flushOk,
     const TVector<ui64>& flushFailed)
 {
+    if (DisabledHosts.Get(route.DestinationHostIndex)) {
+        return;
+    }
+
     for (ui64 lsn: flushOk) {
         auto item = Inflight.GetValue(lsn);
         Y_ABORT_UNLESS(item);
         auto& inflight = item->Value;
 
-        inflight.ConfirmFlush(route);
+        inflight.ConfirmFlush(route.DestinationHostIndex);
     }
 
     for (ui64 lsn: flushFailed) {
@@ -585,7 +591,7 @@ void TBlocksDirtyMap::FlushFinished(
         Y_ABORT_UNLESS(item);
         auto& inflight = item->Value;
 
-        inflight.FlushFailed(route);
+        inflight.FlushFailed(route.DestinationHostIndex);
     }
 }
 
@@ -596,7 +602,10 @@ void TBlocksDirtyMap::EraseFinished(
 {
     for (ui64 lsn: eraseOk) {
         auto item = Inflight.GetValue(lsn);
-        Y_ABORT_UNLESS(item);
+        if (!item) {
+            // The item was deleted when the host was disabled.
+            continue;
+        }
         auto& inflight = item->Value;
 
         if (inflight.ConfirmErase(host)) {
@@ -607,7 +616,10 @@ void TBlocksDirtyMap::EraseFinished(
 
     for (ui64 lsn: eraseFailed) {
         auto item = Inflight.GetValue(lsn);
-        Y_ABORT_UNLESS(item);
+        if (!item) {
+            // The item was deleted when the host was disabled.
+            continue;
+        }
         auto& inflight = item->Value;
 
         inflight.EraseFailed(host);
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 e9afbf2f95f..90da23a359b 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
@@ -17,7 +17,7 @@ constexpr ui64 DefaultVChunkSize = RegionSize / DirectBlockGroupsCount;
 TVChunkConfig MakeTestVChunkConfig()
 {
     return TVChunkConfig::MakeDefault(
-        /*vChunkIndex=*/0,
+        0,   // VChunkIndex
         DirectBlockGroupHostCount,
         DefaultPrimaryCount);
 }
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.cpp
index 50c5ad73bea..2e0b2cc1c14 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.cpp
@@ -191,13 +191,13 @@ THostIndex TInflightInfo::RequestFlush(
     Y_ABORT_UNLESS(false);
 }
 
-void TInflightInfo::ConfirmFlush(THostRoute route)
+void TInflightInfo::ConfirmFlush(THostIndex host)
 {
     Y_ABORT_UNLESS(State == EState::PBufferFlushing);
-    Y_ABORT_UNLESS(FlushRequested.Get(route.DestinationHostIndex));
-    Y_ABORT_UNLESS(!FlushConfirmed.Get(route.DestinationHostIndex));
+    Y_ABORT_UNLESS(FlushRequested.Get(host));
+    Y_ABORT_UNLESS(!FlushConfirmed.Get(host));
 
-    FlushConfirmed.Set(route.DestinationHostIndex);
+    FlushConfirmed.Set(host);
 
     if (FlushDesired == FlushConfirmed) {
         SetState(EState::PBufferFlushed);
@@ -208,13 +208,13 @@ void TInflightInfo::ConfirmFlush(THostRoute route)
     }
 }
 
-void TInflightInfo::FlushFailed(THostRoute route)
+void TInflightInfo::FlushFailed(THostIndex host)
 {
     Y_ABORT_UNLESS(State == EState::PBufferFlushing);
-    Y_ABORT_UNLESS(FlushRequested.Get(route.DestinationHostIndex));
-    Y_ABORT_UNLESS(!FlushConfirmed.Get(route.DestinationHostIndex));
+    Y_ABORT_UNLESS(FlushRequested.Get(host));
+    Y_ABORT_UNLESS(!FlushConfirmed.Get(host));
 
-    FlushRequested.Reset(route.DestinationHostIndex);
+    FlushRequested.Reset(host);
     ReadyQueue->Register(Lsn, IReadyQueue::EQueueType::Flush);
 }
 
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.h
index e90b8feeef8..28b2895469c 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info.h
@@ -148,8 +148,8 @@ public:
     [[nodiscard]] THostIndex RequestFlush(
         THostIndex destination,
         THostMask disabledHosts);
-    void ConfirmFlush(THostRoute route);
-    void FlushFailed(THostRoute route);
+    void ConfirmFlush(THostIndex host);
+    void FlushFailed(THostIndex host);
     [[nodiscard]] THostMask GetRequestedFlushes() const;
 
     void RequestErase(THostIndex host);
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info_ut.cpp
index 113b79e36db..cf46e88e604 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/dirty_map/inflight_info_ut.cpp
@@ -120,17 +120,11 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
             inflightInfo.RequestFlush(THostIndex{2}, THostMask()));
 
         // Confirm flushes
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
         UNIT_ASSERT_VALUES_EQUAL(false, readyQueue.ReadyToErase.contains(123));
-        inflightInfo.ConfirmFlush(
-            {.SourceHostIndex = THostIndex{1},
-             .DestinationHostIndex = THostIndex{1}});
+        inflightInfo.ConfirmFlush(THostIndex{1});
         UNIT_ASSERT_VALUES_EQUAL(false, readyQueue.ReadyToErase.contains(123));
-        inflightInfo.ConfirmFlush(
-            {.SourceHostIndex = THostIndex{2},
-             .DestinationHostIndex = THostIndex{2}});
+        inflightInfo.ConfirmFlush(THostIndex{2});
         UNIT_ASSERT_VALUES_EQUAL(true, readyQueue.ReadyToErase.contains(123));
 
         // Check erase requests
@@ -164,21 +158,15 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
         Y_UNUSED(l);
 
         // Confirm two flushes
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
-        inflightInfo.ConfirmFlush(
-            {.SourceHostIndex = THostIndex{1},
-             .DestinationHostIndex = THostIndex{1}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
+        inflightInfo.ConfirmFlush(THostIndex{1});
 
         // Check lock/unlock PBuffer
         inflightInfo.LockPBuffer();
         UNIT_ASSERT_VALUES_EQUAL(true, readyQueue.ReadyToErase.empty());
 
         // Confirm last flush
-        inflightInfo.ConfirmFlush(
-            {.SourceHostIndex = THostIndex{2},
-             .DestinationHostIndex = THostIndex{2}});
+        inflightInfo.ConfirmFlush(THostIndex{2});
         UNIT_ASSERT_VALUES_EQUAL(true, readyQueue.ReadyToErase.empty());
 
         inflightInfo.UnlockPBuffer();
@@ -220,9 +208,7 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
 
         // When a flush fails, the lsn must be queued for a flush again.
         readyQueue.ReadyToFlush.clear();
-        inflightInfo.FlushFailed(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
+        inflightInfo.FlushFailed(THostIndex{0});
         UNIT_ASSERT_VALUES_EQUAL(true, readyQueue.ReadyToFlush.contains(123));
 
         // Restart flush to host 0
@@ -231,17 +217,11 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
             inflightInfo.RequestFlush(THostIndex{0}, THostMask()));
 
         // Confirm flushes
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
         UNIT_ASSERT_VALUES_EQUAL(false, readyQueue.ReadyToErase.contains(123));
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{1},
-            .DestinationHostIndex = THostIndex{1}});
+        inflightInfo.ConfirmFlush(THostIndex{1});
         UNIT_ASSERT_VALUES_EQUAL(false, readyQueue.ReadyToErase.contains(123));
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{2},
-            .DestinationHostIndex = THostIndex{2}});
+        inflightInfo.ConfirmFlush(THostIndex{2});
         UNIT_ASSERT_VALUES_EQUAL(true, readyQueue.ReadyToErase.contains(123));
 
         // Erase started
@@ -318,15 +298,9 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
             2,
             inflightInfo.RequestFlush(THostIndex{2}, THostMask()));
 
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{1},
-            .DestinationHostIndex = THostIndex{1}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{2},
-            .DestinationHostIndex = THostIndex{2}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
+        inflightInfo.ConfirmFlush(THostIndex{1});
+        inflightInfo.ConfirmFlush(THostIndex{2});
 
         // Before any erase, all written hosts need erasing.
         auto eraseNeeded = inflightInfo.GetEraseNeeded();
@@ -385,15 +359,9 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
             2,
             inflightInfo.RequestFlush(THostIndex{2}, THostMask()));
 
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{1},
-            .DestinationHostIndex = THostIndex{1}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{2},
-            .DestinationHostIndex = THostIndex{2}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
+        inflightInfo.ConfirmFlush(THostIndex{1});
+        inflightInfo.ConfirmFlush(THostIndex{2});
 
         // Request erase for host 0 and then fail it.
         inflightInfo.RequestErase(THostIndex{0});
@@ -563,12 +531,8 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
         Y_UNUSED(l);
 
         // Confirm flush for hosts 0 and 1 only.
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{1},
-            .DestinationHostIndex = THostIndex{1}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
+        inflightInfo.ConfirmFlush(THostIndex{1});
 
         UNIT_ASSERT_VALUES_EQUAL(
             TInflightInfo::EState::PBufferFlushing,
@@ -600,15 +564,9 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
         l = inflightInfo.RequestFlush(THostIndex{1}, THostMask());
         l = inflightInfo.RequestFlush(THostIndex{2}, THostMask());
         Y_UNUSED(l);
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{1},
-            .DestinationHostIndex = THostIndex{1}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{2},
-            .DestinationHostIndex = THostIndex{2}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
+        inflightInfo.ConfirmFlush(THostIndex{1});
+        inflightInfo.ConfirmFlush(THostIndex{2});
 
         // Start erasing hosts 0 and 1, confirm both.
         inflightInfo.RequestErase(THostIndex{0});
@@ -688,12 +646,8 @@ Y_UNIT_TEST_SUITE(TInflightInfoTests)
         l = inflightInfo.RequestFlush(THostIndex{1}, THostMask());
         l = inflightInfo.RequestFlush(THostIndex{2}, THostMask());
         Y_UNUSED(l);
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{0},
-            .DestinationHostIndex = THostIndex{0}});
-        inflightInfo.ConfirmFlush(THostRoute{
-            .SourceHostIndex = THostIndex{1},
-            .DestinationHostIndex = THostIndex{1}});
+        inflightInfo.ConfirmFlush(THostIndex{0});
+        inflightInfo.ConfirmFlush(THostIndex{1});
         UNIT_ASSERT_VALUES_EQUAL(
             TInflightInfo::EState::PBufferFlushing,
             inflightInfo.GetState());
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.cpp
index 69fa7dd04e0..da267134c90 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.cpp
@@ -31,6 +31,8 @@ TFlushRequestExecutor::TFlushRequestExecutor(
     , Span(std::move(span))
     , Route(route)
     , Hint(std::move(hint))
+    , Cooldown(DirectBlockGroup->GetOracle()->GetFlushRequestCooldown(
+          THostMask::MakeFromRoute(Route)))
     , RequestTimeout(DirectBlockGroup->GetOracle()->GetFlushRequestTimeout())
 {
     Y_ABORT_UNLESS(Route.SourceHostIndex != InvalidHostIndex);
@@ -52,21 +54,22 @@ TFlushRequestExecutor::~TFlushRequestExecutor()
 
 void TFlushRequestExecutor::Run()
 {
-    ScheduleRequestTimeout();
-
-    auto future = DirectBlockGroup->SyncWithPBuffer(
-        VChunkConfig.GetVChunkIndex(),
-        Route.SourceHostIndex,
-        Route.DestinationHostIndex,
-        Hint.Segments,
-        Span.GetTraceId());
-    future.Subscribe(
-        [self = shared_from_this()]   //
-        (const NThreading::TFuture<TDBGFlushResponse>& f)
-        {
-            //
-            self->OnFlushResponse(f.GetValue());
-        });
+    if (!Cooldown) {
+        DoRun();
+    } else {
+        LOG_INFO(
+            *ActorSystem,
+            NKikimrServices::NBS_PARTITION,
+            "%s Flush cooldown time %s",
+            LogTitle.GetWithTime().c_str(),
+            FormatDuration(Cooldown).c_str());
+        DirectBlockGroup->Schedule(
+            Cooldown,
+            [self = shared_from_this()]()
+            {
+                self->DoRun();   //
+            });
+    }
 }
 
 TString TFlushRequestExecutor::Print()
@@ -74,6 +77,9 @@ TString TFlushRequestExecutor::Print()
     TStringBuilder result;
     result << LogTitle.GetWithTime();
     result << Hint.DebugPrint(true);
+    if (Cooldown) {
+        result << ",Cooldown=" << FormatDuration(Cooldown);
+    }
     result << (Promise.IsReady() ? ",Replied" : ",NotReplied");
     return result;
 }
@@ -84,6 +90,24 @@ TFlushRequestExecutor::GetFuture() const
     return Promise.GetFuture();
 }
 
+void TFlushRequestExecutor::DoRun()
+{
+    ScheduleRequestTimeout();
+
+    auto future = DirectBlockGroup->SyncWithPBuffer(
+        VChunkConfig.GetVChunkIndex(),
+        Route.SourceHostIndex,
+        Route.DestinationHostIndex,
+        Hint.Segments,
+        Span.GetTraceId());
+    future.Subscribe(
+        [self = shared_from_this()]   //
+        (const NThreading::TFuture<TDBGFlushResponse>& f)
+        {
+            self->OnFlushResponse(f.GetValue());   //
+        });
+}
+
 void TFlushRequestExecutor::OnFlushResponse(const TDBGFlushResponse& response)
 {
     Y_ABORT_UNLESS(Hint.Segments.size() == response.Errors.size());
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.h
index 62b40a1b702..9f007e59f3d 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request.h
@@ -43,6 +43,7 @@ public:
     NThreading::TFuture<TResponse> GetFuture() const;
 
 private:
+    void DoRun();
     void SendFlushRequest(THostIndex host);
     void OnFlushResponse(const TDBGFlushResponse& response);
     void Reply(TVector<ui64> flushOk, TVector<ui64> flushFailed);
@@ -57,6 +58,7 @@ private:
     const NWilson::TSpan Span;
     const THostRoute Route;
     const TFlushHint Hint;
+    const TDuration Cooldown;
     const TDuration RequestTimeout;
 
     NThreading::TPromise<TResponse> Promise =
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request_ut.cpp
index 53580240731..8a1855a9ecb 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/flush_request_ut.cpp
@@ -180,6 +180,61 @@ Y_UNIT_TEST_SUITE(TFlushRequestTest)
         UNIT_ASSERT_VALUES_EQUAL(43, response.FlushFailed[1]);
         UNIT_ASSERT_EQUAL_C(route, response.Route, response.Route.DebugPrint());
     }
+
+    Y_UNIT_TEST_F(CooldownDelaysFlushRequest, TBaseFixture)
+    {
+        Init();
+
+        // When the oracle reports a non-zero cooldown, the flush request must
+        // not be issued immediately. Instead it is scheduled after the
+        // cooldown delay.
+        const auto cooldown = TDuration::MilliSeconds(30);
+        DirectBlockGroup->Oracle.FlushRequestCooldown = cooldown;
+        DirectBlockGroup->Oracle.FlushRequestTimeout = TDuration::Seconds(5);
+
+        THostRoute route{.SourceHostIndex = 1, .DestinationHostIndex = 2};
+        TFlushHint hint;
+        hint.Segments.push_back(TPBufferSegment{
+            .Lsn = 42,
+            .Range = TBlockRange64::WithLength(10, 3)});
+
+        auto flushRequest = std::make_shared<TFlushRequestExecutor>(
+            Runtime->GetActorSystem(0),
+            LogTitle,
+            VChunkConfig,
+            DirectBlockGroup,
+            route,
+            std::move(hint),
+            NWilson::TSpan());
+
+        auto future = flushRequest->GetFuture();
+        flushRequest->Run();
+
+        // No flush request is issued yet - only the cooldown was scheduled.
+        UNIT_ASSERT_VALUES_EQUAL(false, future.HasValue());
+        UNIT_ASSERT_VALUES_EQUAL(0, FlushPromises.size());
+        UNIT_ASSERT_VALUES_EQUAL(1, ScheduledTasks.size());
+        UNIT_ASSERT_VALUES_EQUAL(cooldown, ScheduledTasks[0].Delay);
+
+        // Fire the cooldown timer - the actual flush request is now issued.
+        ScheduledTasks[0].Callback();
+
+        UNIT_ASSERT_VALUES_EQUAL(1, FlushPromises.size());
+        // A request timeout is scheduled once the flush actually starts.
+        UNIT_ASSERT_VALUES_EQUAL(2, ScheduledTasks.size());
+        UNIT_ASSERT_VALUES_EQUAL(
+            DirectBlockGroup->Oracle.FlushRequestTimeout,
+            ScheduledTasks[1].Delay);
+
+        SetFlushResult(TDBGFlushResponse{.Errors{MakeError(S_OK)}}, false);
+
+        UNIT_ASSERT_VALUES_EQUAL(true, future.HasValue());
+        const auto& response = future.GetValue();
+        UNIT_ASSERT_VALUES_EQUAL(1, response.FlushOk.size());
+        UNIT_ASSERT_VALUES_EQUAL(42, response.FlushOk[0]);
+        UNIT_ASSERT_VALUES_EQUAL(0, response.FlushFailed.size());
+        UNIT_ASSERT_EQUAL_C(route, response.Route, response.Route.DebugPrint());
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.cpp
index 7537786d26f..6ff06638028 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.cpp
@@ -36,6 +36,15 @@ THostMask THostMask::MakeAll(size_t hostCount)
     return THostMask((ui32(1) << hostCount) - 1);
 }
 
+// static
+THostMask THostMask::MakeFromRoute(const THostRoute& route)
+{
+    THostMask result;
+    result.Set(route.SourceHostIndex);
+    result.Set(route.DestinationHostIndex);
+    return result;
+}
+
 void THostMask::Set(THostIndex host)
 {
     Y_ABORT_UNLESS(host < MaxHostCount);
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.h
index dc2c099e81b..43329162728 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask.h
@@ -34,6 +34,7 @@ public:
     static THostMask MakeEmpty();
     static THostMask MakeOne(THostIndex host);
     static THostMask MakeAll(size_t hostCount);
+    static THostMask MakeFromRoute(const THostRoute& route);
 
     void Set(THostIndex host);
     void Reset(THostIndex host);
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask_ut.cpp
index f2a8b9a0f56..eec319ae377 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_mask_ut.cpp
@@ -44,6 +44,23 @@ Y_UNIT_TEST_SUITE(THostMaskTest)
         UNIT_ASSERT(mask.Get(7));
     }
 
+    Y_UNIT_TEST(ShouldMakeFromRoute)
+    {
+        // A route with distinct source and destination sets both bits.
+        auto mask = THostMask::MakeFromRoute(
+            THostRoute{.SourceHostIndex = 1, .DestinationHostIndex = 4});
+        UNIT_ASSERT_VALUES_EQUAL(2u, mask.Count());
+        UNIT_ASSERT(mask.Get(1));
+        UNIT_ASSERT(mask.Get(4));
+        UNIT_ASSERT(!mask.Get(0));
+
+        // A route where source equals destination collapses to a single bit.
+        auto sameHost = THostMask::MakeFromRoute(
+            THostRoute{.SourceHostIndex = 2, .DestinationHostIndex = 2});
+        UNIT_ASSERT_VALUES_EQUAL(1u, sameHost.Count());
+        UNIT_ASSERT(sameHost.Get(2));
+    }
+
     Y_UNIT_TEST(ShouldDoLogicalOps)
     {
         auto first = THostMask::MakeOne(0).Include(THostMask::MakeOne(1));
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.cpp
index 6f17f870794..c442a6fa92e 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.cpp
@@ -78,6 +78,11 @@ size_t THostStat::GetConsecutiveSuccessCount() const
     return ConsecutiveSuccessCount;
 }
 
+size_t THostStat::GetConsecutiveErrorCount() const
+{
+    return ConsecutiveErrorCount;
+}
+
 size_t THostStat::InflightCount(EOperation operation) const
 {
     return InflightByOperation[static_cast<size_t>(operation)];
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.h
index 70ba3ad9fe1..a407b69dec5 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat.h
@@ -57,6 +57,9 @@ public:
     // Number of consecutive successful completions since the last error
     // (reset to 0 on the first error after a success streak).
     [[nodiscard]] size_t GetConsecutiveSuccessCount() const;
+    // Number of consecutive failed completions since the last success
+    // (reset to 0 on the first success after a error streak).
+    [[nodiscard]] size_t GetConsecutiveErrorCount() const;
 
     // Number of currently inflight requests of a given operation type for
     // this host (i.e. OnRequest calls without a matching OnSuccess/OnError).
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat_ut.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat_ut.cpp
index e409f2201ef..a0d965d5d18 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat_ut.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/model/host_stat_ut.cpp
@@ -190,6 +190,37 @@ Y_UNIT_TEST_SUITE(THostStatTest)
         UNIT_ASSERT_VALUES_EQUAL(1, errorsInfo.ConsecutiveErrorCount);
     }
 
+    Y_UNIT_TEST(ErrorCountGetterReflectsConsecutiveErrors)
+    {
+        THostStat stat;
+        TInstant now = TInstant::Now();
+
+        // Initially there are no consecutive errors.
+        UNIT_ASSERT_VALUES_EQUAL(0, stat.GetConsecutiveErrorCount());
+
+        // Each error increments the consecutive error count.
+        stat.OnRequest(EOperation::WriteToPBuffer);
+        stat.OnError(now, EOperation::WriteToPBuffer);
+        UNIT_ASSERT_VALUES_EQUAL(1, stat.GetConsecutiveErrorCount());
+
+        stat.OnRequest(EOperation::WriteToPBuffer);
+        stat.OnError(now, EOperation::WriteToPBuffer);
+        UNIT_ASSERT_VALUES_EQUAL(2, stat.GetConsecutiveErrorCount());
+
+        // A success resets the consecutive error count back to zero.
+        stat.OnRequest(EOperation::WriteToPBuffer);
+        stat.OnSuccess(now, TDuration(), EOperation::WriteToPBuffer);
+        UNIT_ASSERT_VALUES_EQUAL(0, stat.GetConsecutiveErrorCount());
+
+        // OnCancelled does not affect the consecutive error count.
+        stat.OnRequest(EOperation::WriteToPBuffer);
+        stat.OnError(now, EOperation::WriteToPBuffer);
+        UNIT_ASSERT_VALUES_EQUAL(1, stat.GetConsecutiveErrorCount());
+        stat.OnRequest(EOperation::WriteToPBuffer);
+        stat.OnCancelled(now, EOperation::WriteToPBuffer);
+        UNIT_ASSERT_VALUES_EQUAL(1, stat.GetConsecutiveErrorCount());
+    }
+
     Y_UNIT_TEST(SuccessCountInitiallyZero)
     {
         THostStat stat;
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 a277aa74d7c..81a2db7ca01 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
@@ -14,8 +14,13 @@ namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect {
 
 namespace {
 
+////////////////////////////////////////////////////////////////////////////////
+
 constexpr TDuration MinReconnectDelay = TDuration::MilliSeconds(20);
 constexpr TDuration MaxReconnectDelay = TDuration::Seconds(10);
+constexpr TDuration FlushRequestCooldownPenalty = TDuration::MilliSeconds(10);
+
+////////////////////////////////////////////////////////////////////////////////
 
 TDuration GetFromConfig(ui64 milliseconds, TDuration defaultValue)
 {
@@ -316,6 +321,11 @@ TDuration TOracle::GetReadRequestTimeout() const
     return DefaultReadRequestTimeout;
 }
 
+EWriteMode TOracle::GetWriteMode() const
+{
+    return DefaultWriteMode;
+}
+
 TDuration TOracle::GetWriteHedgingDelay(THostMask hosts, bool indirect) const
 {
     TDuration result = GetTimePredictor(
@@ -335,6 +345,25 @@ TDuration TOracle::GetIndirectWriteReplyTimeout() const
     return DefaultIndirectWriteReplyTimeout;
 }
 
+TDuration TOracle::GetFlushRequestCooldown(THostMask hosts) const
+{
+    auto cooldown = [&](THostIndex host) -> TDuration
+    {
+        const size_t errorCount =
+            HostStatistics[host].GetConsecutiveErrorCount();
+        if (!errorCount) {
+            return TDuration::Zero();
+        }
+        return errorCount * FlushRequestCooldownPenalty;
+    };
+
+    TDuration result;
+    for (auto host: hosts) {
+        result = Max(result, cooldown(host));
+    }
+    return Min(result, MaxReconnectDelay);
+}
+
 TDuration TOracle::GetFlushRequestTimeout() const
 {
     return DefaultFlushRequestTimeout;
@@ -345,11 +374,6 @@ TDuration TOracle::GetEraseRequestTimeout() const
     return DefaultEraseRequestTimeout;
 }
 
-EWriteMode TOracle::GetWriteMode() const
-{
-    return DefaultWriteMode;
-}
-
 const THostStat& TOracle::GetHostStatistics(THostIndex hostIndex) const
 {
     return HostStatistics[hostIndex];
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 f2100759169..65fb5ab8a11 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
@@ -66,14 +66,19 @@ public:
         THostIndex host,
         EDataLocation dataLocation) const = 0;
     [[nodiscard]] virtual TDuration GetReadRequestTimeout() const = 0;
+
+    [[nodiscard]] virtual EWriteMode GetWriteMode() const = 0;
     [[nodiscard]] virtual TDuration GetWriteHedgingDelay(
         THostMask hosts,
         bool indirect) const = 0;
     [[nodiscard]] virtual TDuration GetWriteRequestTimeout() const = 0;
     [[nodiscard]] virtual TDuration GetIndirectWriteReplyTimeout() const = 0;
+
+    [[nodiscard]] virtual TDuration GetFlushRequestCooldown(
+        THostMask hosts) const = 0;
     [[nodiscard]] virtual TDuration GetFlushRequestTimeout() const = 0;
+
     [[nodiscard]] virtual TDuration GetEraseRequestTimeout() const = 0;
-    [[nodiscard]] virtual EWriteMode GetWriteMode() const = 0;
 
     [[nodiscard]] virtual const THostStat& GetHostStatistics(
         THostIndex hostIndex) const = 0;
@@ -92,6 +97,7 @@ public:
 
     void Think(TInstant now);
 
+    // IOracle implementation
     void OnRequestStarted(
         THostIndex hostIndex,
         EOperation operation,
@@ -123,14 +129,19 @@ public:
         THostIndex host,
         EDataLocation dataLocation) const override;
     [[nodiscard]] TDuration GetReadRequestTimeout() const override;
+
+    [[nodiscard]] EWriteMode GetWriteMode() const override;
     [[nodiscard]] TDuration GetWriteHedgingDelay(
         THostMask hosts,
         bool indirect) const override;
     [[nodiscard]] TDuration GetWriteRequestTimeout() const override;
     [[nodiscard]] TDuration GetIndirectWriteReplyTimeout() const override;
+
+    [[nodiscard]] TDuration GetFlushRequestCooldown(
+        THostMask hosts) const override;
     [[nodiscard]] TDuration GetFlushRequestTimeout() const override;
+
     [[nodiscard]] TDuration GetEraseRequestTimeout() const override;
-    [[nodiscard]] EWriteMode GetWriteMode() const override;
 
     [[nodiscard]] const THostStat& GetHostStatistics(
         THostIndex hostIndex) 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 5955cd35d4d..aef44ab809d 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
@@ -541,6 +541,121 @@ Y_UNIT_TEST_SUITE(TOracle)
             TDuration::MilliSeconds(300),
             oracle.GetWriteHedgingDelay(THostMask::MakeOne(0), true));
     }
+
+    Y_UNIT_TEST(GetFlushRequestCooldownReturnsZeroWithoutErrors)
+    {
+        NProto::TStorageServiceConfig rawConfig;
+        auto storageConfig = std::make_shared<TStorageConfig>(rawConfig);
+
+        TOracle oracle(storageConfig, nullptr);
+
+        // No errors recorded on any host -> no cooldown.
+        UNIT_ASSERT_VALUES_EQUAL(
+            TDuration::Zero(),
+            oracle.GetFlushRequestCooldown(THostMask::MakeAll(5)));
+    }
+
+    Y_UNIT_TEST(GetFlushRequestCooldownScalesWithConsecutiveErrors)
+    {
+        NProto::TStorageServiceConfig rawConfig;
+        auto storageConfig = std::make_shared<TStorageConfig>(rawConfig);
+
+        TOracle oracle(storageConfig, nullptr);
+        const auto now = TInstant::Now();
+
+        // Each consecutive error adds a 10ms penalty on host 0.
+        for (size_t errorCount = 1; errorCount <= 3; ++errorCount) {
+            oracle.OnRequestStarted(0, EOperation::WriteToPBuffer, now);
+            oracle.OnRequestFailed(0, EOperation::WriteToPBuffer, now);
+
+            UNIT_ASSERT_VALUES_EQUAL(
+                TDuration::MilliSeconds(10 * errorCount),
+                oracle.GetFlushRequestCooldown(THostMask::MakeOne(0)));
+        }
+
+        // A single success resets the consecutive error count back to zero.
+        oracle.OnRequestStarted(0, EOperation::WriteToPBuffer, now);
+        oracle.OnRequestSucceeded(
+            0,
+            EOperation::WriteToPBuffer,
+            now,
+            TDuration());
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            TDuration::Zero(),
+            oracle.GetFlushRequestCooldown(THostMask::MakeOne(0)));
+    }
+
+    Y_UNIT_TEST(GetFlushRequestCooldownReturnsMaxAcrossHosts)
+    {
+        NProto::TStorageServiceConfig rawConfig;
+        auto storageConfig = std::make_shared<TStorageConfig>(rawConfig);
+
+        TOracle oracle(storageConfig, nullptr);
+        const auto now = TInstant::Now();
+
+        // Host 1: two consecutive errors -> 20ms.
+        for (size_t i = 0; i < 2; ++i) {
+            oracle.OnRequestStarted(1, EOperation::WriteToPBuffer, now);
+            oracle.OnRequestFailed(1, EOperation::WriteToPBuffer, now);
+        }
+
+        // Host 3: four consecutive errors -> 40ms.
+        for (size_t i = 0; i < 4; ++i) {
+            oracle.OnRequestStarted(3, EOperation::WriteToPBuffer, now);
+            oracle.OnRequestFailed(3, EOperation::WriteToPBuffer, now);
+        }
+
+        // A mask covering both hosts must return the maximum cooldown.
+        UNIT_ASSERT_VALUES_EQUAL(
+            TDuration::MilliSeconds(40),
+            oracle.GetFlushRequestCooldown(
+                THostMask::MakeOne(1).Include(THostMask::MakeOne(3))));
+
+        // A mask restricted to the less-penalized host returns its own value.
+        UNIT_ASSERT_VALUES_EQUAL(
+            TDuration::MilliSeconds(20),
+            oracle.GetFlushRequestCooldown(THostMask::MakeOne(1)));
+
+        // A mask covering only error-free hosts returns zero.
+        UNIT_ASSERT_VALUES_EQUAL(
+            TDuration::Zero(),
+            oracle.GetFlushRequestCooldown(
+                THostMask::MakeOne(0).Include(THostMask::MakeOne(4))));
+    }
+
+    Y_UNIT_TEST(GetFlushRequestCooldownIsClampedToMaxReconnectDelay)
+    {
+        NProto::TStorageServiceConfig rawConfig;
+        auto storageConfig = std::make_shared<TStorageConfig>(rawConfig);
+
+        TOracle oracle(storageConfig, nullptr);
+        const auto now = TInstant::Now();
+
+        // The cooldown grows by 10ms per consecutive error and is capped at
+        // MaxReconnectDelay (10s). 10s / 10ms == 1000 errors reach the cap
+        // exactly; any additional error must not exceed it.
+        const auto maxReconnectDelay = TDuration::Seconds(10);
+        for (size_t i = 0; i < 1000; ++i) {
+            oracle.OnRequestStarted(0, EOperation::WriteToPBuffer, now);
+            oracle.OnRequestFailed(0, EOperation::WriteToPBuffer, now);
+        }
+
+        // Exactly at the cap.
+        UNIT_ASSERT_VALUES_EQUAL(
+            maxReconnectDelay,
+            oracle.GetFlushRequestCooldown(THostMask::MakeOne(0)));
+
+        // Far beyond the cap - still clamped.
+        for (size_t i = 0; i < 500; ++i) {
+            oracle.OnRequestStarted(0, EOperation::WriteToPBuffer, now);
+            oracle.OnRequestFailed(0, EOperation::WriteToPBuffer, now);
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(
+            maxReconnectDelay,
+            oracle.GetFlushRequestCooldown(THostMask::MakeOne(0)));
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
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 616189f8841..08ff001725c 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
@@ -89,6 +89,16 @@ ui32 TVChunkConfig::GetVChunkIndex() const
     return VChunkIndex;
 }
 
+void TVChunkConfig::SetDBGIndex(ui32 dbgIndex)
+{
+    DBGIndex = dbgIndex;
+}
+
+ui32 TVChunkConfig::GetDBGIndex() const
+{
+    return DBGIndex;
+}
+
 void TVChunkConfig::EnableHost(THostIndex hostIndex)
 {
     EnabledHosts.Set(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 6a0e5374c3c..4069fee3da6 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
@@ -25,6 +25,9 @@ public:
     [[nodiscard]] size_t GetHostCount() const;
     [[nodiscard]] ui32 GetVChunkIndex() const;
 
+    void SetDBGIndex(ui32 dbgIndex);
+    [[nodiscard]] ui32 GetDBGIndex() const;
+
     // Enables the host to work. If the total count of ddisks is not enough to
     // reach the quorum, then can make a promotion for ddisk.
     void EnableHost(THostIndex hostIndex);
@@ -73,6 +76,7 @@ public:
 
 private:
     size_t HostCount = 0;
+    ui32 DBGIndex = 0;
     ui32 VChunkIndex = 0;
     THostRoles PBufferHosts;
     THostRoles DDiskHosts;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.cpp
index 7bc57d17bee..d0c147f6ecc 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.cpp
@@ -12,6 +12,16 @@
 
 namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect {
 
+namespace {
+
+////////////////////////////////////////////////////////////////////////////////
+
+constexpr auto SlowRequestTime = TDuration::Seconds(1);
+
+////////////////////////////////////////////////////////////////////////////////
+
+}   // namespace
+
 ////////////////////////////////////////////////////////////////////////////////
 
 TReadSingleLocationRequestExecutor::TReadSingleLocationRequestExecutor(
@@ -56,6 +66,7 @@ TReadSingleLocationRequestExecutor::~TReadSingleLocationRequestExecutor()
 
 void TReadSingleLocationRequestExecutor::Run()
 {
+    StartAt = TInstant::Now();
     ScheduleRequestTimeout();
 
     StartReading();
@@ -64,11 +75,8 @@ void TReadSingleLocationRequestExecutor::Run()
 TString TReadSingleLocationRequestExecutor::Print()
 {
     TStringBuilder result;
-    result << LogTitle.GetWithTime();
-    result << "c:" << ReadHint.HostMask.Print();
-    result << ",r:" << Requested.Print();
-    result << ",f:" << Failed.Print();
-    result << (Promise.IsReady() ? ",Replied" : ",NotReplied");
+    result << LogTitle.GetWithTime() << " " << ExtendedDebugState() << " "
+           << (Promise.IsReady() ? ",Replied" : ",NotReplied");
 
     return result;
 }
@@ -90,18 +98,14 @@ void TReadSingleLocationRequestExecutor::StartReading()
     auto host = candidates.First();
     if (!host) {
         if (Requested == Failed) {
-            Reply(MakeError(
-                E_REJECTED,
-                TStringBuilder() << "Can't read. r:" << Requested.Print()
-                                 << ",f:" << Failed.Print()));
+            Reply(MakeError(E_REJECTED, ExtendedDebugState()));
         } else {
             LOG_DEBUG(
                 *ActorSystem,
                 NKikimrServices::NBS_PARTITION,
-                "%s Read started from all available hosts. r:%s,f:%s",
+                "%s Read started from all available hosts. %s",
                 LogTitle.GetWithTime().c_str(),
-                Requested.Print().c_str(),
-                Failed.Print().c_str());
+                ExtendedDebugState().c_str());
         }
         return;
     }
@@ -126,11 +130,9 @@ void TReadSingleLocationRequestExecutor::StartReading()
         PrintHostIndex(*host).c_str(),
         tryCount);
 
-    auto onReadResponse = [self = shared_from_this(), host = *host]   //
-        (const NThreading::TFuture<TDBGReadBlocksResponse>& f)
-    {
-        self->OnReadResponse(host, f.GetValue());
-    };
+    ScheduleHedging(DirectBlockGroup->GetOracle()->GetReadHedgingDelay(
+        *host,
+        ReadHint.Lsn == 0 ? EDataLocation::DDisk : EDataLocation::PBuffer));
 
     auto future = fromDDisk ? DirectBlockGroup->ReadBlocksFromDDisk(
                                   VChunkConfig.GetVChunkIndex(),
@@ -145,11 +147,12 @@ void TReadSingleLocationRequestExecutor::StartReading()
                                   ReadHint.VChunkRange,
                                   Request->Sglist,
                                   TraceId);
-    future.Subscribe(std::move(onReadResponse));
-
-    ScheduleHedging(DirectBlockGroup->GetOracle()->GetReadHedgingDelay(
-        *host,
-        ReadHint.Lsn == 0 ? EDataLocation::DDisk : EDataLocation::PBuffer));
+    future.Subscribe(
+        [self = shared_from_this(), host = *host]   //
+        (const NThreading::TFuture<TDBGReadBlocksResponse>& f)
+        {
+            self->OnReadResponse(host, f.GetValue());   //
+        });
 }
 
 void TReadSingleLocationRequestExecutor::OnReadResponse(
@@ -165,12 +168,11 @@ void TReadSingleLocationRequestExecutor::OnReadResponse(
     LOG_WARN(
         *ActorSystem,
         NKikimrServices::NBS_PARTITION,
-        "%s %s: %s, r:%s,f:%s",
+        "%s %s: %s, %s",
         LogTitle.GetWithTime().c_str(),
         PrintHostIndex(host).c_str(),
         FormatError(response.Error).c_str(),
-        Requested.Print().c_str(),
-        Failed.Print().c_str());
+        ExtendedDebugState().c_str());
 
     StartReading();
 }
@@ -181,19 +183,31 @@ void TReadSingleLocationRequestExecutor::Reply(NProto::TError error)
         return;
     }
 
+    const auto duration = TInstant::Now() - StartAt;
+    if (duration > SlowRequestTime) {
+        LOG_INFO(
+            *ActorSystem,
+            NKikimrServices::NBS_PARTITION,
+            "%s [?] Slow request %s",
+            LogTitle.GetWithTime().c_str(),
+            ExtendedDebugState().c_str());
+    }
+
     if (HasError(error)) {
         LOG_ERROR(
             *ActorSystem,
             NKikimrServices::NBS_PARTITION,
-            "%s [!] Reply error: %s",
+            "%s [!] Reply error: %s %s",
             LogTitle.GetWithTime().c_str(),
-            FormatError(error).c_str());
+            FormatError(error).c_str(),
+            ExtendedDebugState().c_str());
     } else {
         LOG_DEBUG(
             *ActorSystem,
             NKikimrServices::NBS_PARTITION,
-            "%s Reply OK",
-            LogTitle.GetWithTime().c_str());
+            "%s Reply OK %s",
+            LogTitle.GetWithTime().c_str(),
+            ExtendedDebugState().c_str());
     }
 
     ReadHint.Lock.Disarm();
@@ -211,8 +225,9 @@ void TReadSingleLocationRequestExecutor::ScheduleHedging(TDuration hedgingDelay)
     LOG_DEBUG(
         *ActorSystem,
         NKikimrServices::NBS_PARTITION,
-        "%s Schedule OnHedgingTimeout %s",
+        "%s Schedule OnHedgingTimeout %s %s",
         LogTitle.GetWithTime().c_str(),
+        ExtendedDebugState().c_str(),
         FormatDuration(hedgingDelay).c_str());
 
     DirectBlockGroup->Schedule(
@@ -259,7 +274,7 @@ void TReadSingleLocationRequestExecutor::OnHedgingTimeout()
         NKikimrServices::NBS_PARTITION,
         "%s OnHedgingTimeout %s",
         LogTitle.GetWithTime().c_str(),
-        Print().c_str());
+        ExtendedDebugState().c_str());
 
     const bool allRetriesAreSpent = ReadHint.HostMask == Requested;
     if (!allRetriesAreSpent) {
@@ -276,12 +291,22 @@ void TReadSingleLocationRequestExecutor::OnRequestTimeout()
     LOG_WARN(
         *ActorSystem,
         NKikimrServices::NBS_PARTITION,
-        "%s OnRequestTimeout.",
-        LogTitle.GetWithTime().c_str());
+        "%s OnRequestTimeout. %s",
+        LogTitle.GetWithTime().c_str(),
+        ExtendedDebugState().c_str());
 
     Reply(MakeError(E_TIMEOUT, "Request timeout"));
 }
 
+TString TReadSingleLocationRequestExecutor::ExtendedDebugState() const
+{
+    TStringBuilder result;
+    result << "a:" << ReadHint.HostMask.Print();
+    result << ",r:" << Requested.Print();
+    result << ",f:" << Failed.Print();
+    return result;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 }   // namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.h
index 9546247e43b..292da6385e1 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/read_request_single_location.h
@@ -54,6 +54,8 @@ private:
     void OnHedgingTimeout();
     void OnRequestTimeout();
 
+    TString ExtendedDebugState() const;
+
     NActors::TActorSystem const* ActorSystem;
     const TChildLogTitle LogTitle;
     const TVChunkConfig VChunkConfig;
@@ -63,6 +65,7 @@ private:
     const NWilson::TTraceId TraceId;
     const TDuration RequestTimeout;
 
+    TInstant StartAt;
     TReadRangeHint ReadHint;
     THostMask Requested;
     THostMask Failed;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/region.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/region.cpp
index 0a8c5843ecf..209953e944b 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/region.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/region.cpp
@@ -44,11 +44,12 @@ TRegion::TRegion(
             counters->GetSubgroup("vchunk", ToString(vChunkIndex));
 
         const auto* persisted = vChunkConfigs.FindPtr(vChunkIndex);
-        const auto vChunkConfig = persisted ? *persisted
-                                            : TVChunkConfig::MakeDefault(
-                                                  vChunkIndex,
-                                                  DirectBlockGroupHostCount,
-                                                  DefaultPrimaryCount);
+        auto vChunkConfig = persisted ? *persisted
+                                      : TVChunkConfig::MakeDefault(
+                                            vChunkIndex,
+                                            DirectBlockGroupHostCount,
+                                            DefaultPrimaryCount);
+        vChunkConfig.SetDBGIndex(dbgIndex);
         Y_ABORT_UNLESS(vChunkConfig.IsValid());
         Y_ABORT_UNLESS(vChunkConfig.GetVChunkIndex() == vChunkIndex);
 
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 dd780c78830..6622d221149 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
@@ -40,7 +40,10 @@ TVChunk::TVChunk(
     , BlockSize(DefaultBlockSize)
     , BlocksCount(vChunkSize / BlockSize)
     , SyncRequestsBatchSize(syncRequestsBatchSize)
-    , LogTitle{GetCycleCount(), TLogTitle::TVChunk{.VChunkIndex = vChunkConfig.GetVChunkIndex()}}
+    , LogTitle{GetCycleCount(), TLogTitle::TVChunk{
+        .DBGIndex = vChunkConfig.GetDBGIndex(),
+        .VChunkIndex = vChunkConfig.GetVChunkIndex()
+     }}
     , VChunkConfig(vChunkConfig)
     , BlocksDirtyMap(VChunkConfig, BlockSize, BlocksCount)
     , Counters(std::move(counters))
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.cpp b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.cpp
index ccb6cdfcce8..46edb5ba795 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.cpp
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.cpp
@@ -13,6 +13,16 @@
 
 namespace NYdb::NBS::NBlockStore::NStorage::NPartitionDirect {
 
+namespace {
+
+////////////////////////////////////////////////////////////////////////////////
+
+constexpr auto SlowRequestTime = TDuration::Seconds(1);
+
+////////////////////////////////////////////////////////////////////////////////
+
+}   // namespace
+
 ////////////////////////////////////////////////////////////////////////////////
 
 TWriteRequestExecutor::TWriteRequestExecutor(
@@ -53,6 +63,7 @@ TWriteRequestExecutor::~TWriteRequestExecutor()
 
 void TWriteRequestExecutor::Run()
 {
+    StartAt = TInstant::Now();
     Bundle->GetSpan().Event("Run");
 
     const auto hosts = VChunkConfig.GetDesiredPBuffers();
@@ -366,6 +377,17 @@ void TWriteRequestExecutor::Reply(NProto::TError error)
     Y_ABORT_IF(IsReplied, "TWriteRequestExecutor::Reply called twice");
     IsReplied = true;
 
+    const auto duration = TInstant::Now() - StartAt;
+    if (duration > SlowRequestTime) {
+        LOG_INFO(
+            *ActorSystem,
+            NKikimrServices::NBS_PARTITION,
+            "%s [?] Slow request %s %s",
+            LogTitle.GetWithTime().c_str(),
+            ExtendedDebugState().c_str(),
+            FormatDuration(duration).c_str());
+    }
+
     if (HasError(error)) {
         LOG_ERROR(
             *ActorSystem,
@@ -378,8 +400,9 @@ void TWriteRequestExecutor::Reply(NProto::TError error)
         LOG_DEBUG(
             *ActorSystem,
             NKikimrServices::NBS_PARTITION,
-            "%s Reply OK.",
-            LogTitle.GetWithTime().c_str());
+            "%s Reply OK. %s",
+            LogTitle.GetWithTime().c_str(),
+            ExtendedDebugState().c_str());
     }
 
     Bundle->Reply(
@@ -414,8 +437,9 @@ void TWriteRequestExecutor::ScheduleHedging(TDuration hedgingDelay)
     LOG_DEBUG(
         *ActorSystem,
         NKikimrServices::NBS_PARTITION,
-        "%s Schedule OnHedgingTimeout() %s",
+        "%s Schedule OnHedgingTimeout %s %s",
         LogTitle.GetWithTime().c_str(),
+        ExtendedDebugState().c_str(),
         FormatDuration(hedgingDelay).c_str());
 
     DirectBlockGroup->Schedule(
@@ -477,8 +501,9 @@ void TWriteRequestExecutor::OnRequestTimeout()
     LOG_WARN(
         *ActorSystem,
         NKikimrServices::NBS_PARTITION,
-        "%s Request timeout.",
-        LogTitle.GetWithTime().c_str());
+        "%s Request timeout. %s",
+        LogTitle.GetWithTime().c_str(),
+        ExtendedDebugState().c_str());
 
     ReplyOrNotifyBelated(MakeError(E_TIMEOUT, "Write request timeout"), {});
 }
@@ -514,8 +539,8 @@ THostMask TWriteRequestExecutor::GetRunningDirectWrites() const
 TString TWriteRequestExecutor::ExtendedDebugState() const
 {
     TStringBuilder result;
-    result << "dr:" << RequestedDirectWrites.Print();
-    result << " ir:" << IndirectCoordinator.Print()
+    result << "d:" << RequestedDirectWrites.Print();
+    result << " i:" << IndirectCoordinator.Print()
            << RequestedIndirectWrites.Print();
     result << " c:" << CompletedWrites.Print();
     result << " f:" << FailedWrites.Print();
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.h b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.h
index c12d545375c..7cd6e4902c6 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/partition_direct/write_request.h
@@ -79,6 +79,7 @@ private:
     const TDuration RequestTimeout;
     const TDuration IndirectWriteReplyTimeout;
 
+    TInstant StartAt;
     THostMask IndirectCoordinator;
     THostMask RequestedIndirectWrites;
     THostMask RequestedDirectWrites;
diff --git a/ydb/core/nbs/cloud/blockstore/libs/storage/storage_transport/ic_storage_transport_actor.h b/ydb/core/nbs/cloud/blockstore/libs/storage/storage_transport/ic_storage_transport_actor.h
index 282d7688b1c..ca854100fbd 100644
--- a/ydb/core/nbs/cloud/blockstore/libs/storage/storage_transport/ic_storage_transport_actor.h
+++ b/ydb/core/nbs/cloud/blockstore/libs/storage/storage_transport/ic_storage_transport_actor.h
@@ -110,7 +110,6 @@ private:
     void HandleBatchErasePersistentBuffer(
         const TEvTransportPrivate::TEvBatchEraseFromPBuffer::TPtr& ev,
         const NActors::TActorContext& ctx);
-
     void HandleBarrierErasePersistentBuffer(
         const TEvTransportPrivate::TEvBarrierEraseFromPBuffer::TPtr& ev,
         const NActors::TActorContext& ctx);