NPB free chunks (#45916)

13 files changed, 561 insertions, 0 deletions

diff --git a/ydb/core/blobstorage/ddisk/ddisk.h b/ydb/core/blobstorage/ddisk/ddisk.h
index f5de90607ca..7b5297d7602 100644
--- a/ydb/core/blobstorage/ddisk/ddisk.h
+++ b/ydb/core/blobstorage/ddisk/ddisk.h
@@ -270,6 +270,11 @@ struct TPersistentBufferFormat {
     // Allocate a new chunk proactively when free space drops below this percentage
     // of the currently owned capacity. 0 disables proactive allocation.
     ui32 PreallocateFreeSpaceThresholdPercent = 10;
+    // Deallocate a chunk proactively when free space is over this percentage
+    // of the currently owned capacity. 100% disables proactive deallocation.
+    ui32 DeallocateFreeSpaceThresholdPercent = 90;
+    // Deallocate a chunk proactively when it has been freed for this many seconds.
+    ui32 DeallocateThresholdSeconds = 30;
 };
 
 #define DECLARE_DDISK_EVENT(NAME) \
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
index 64e36f15fbf..8f0eda835c9 100644
--- a/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
+++ b/ydb/core/blobstorage/ddisk/ddisk_actor.cpp
@@ -282,6 +282,7 @@ namespace {
             hFunc(TEvSync, handleQuery)
             hFunc(TEvDeleteTabletChunks, handleQuery)
             hFunc(TEvPrivate::TEvIssuePersistentBufferChunkAllocation, Handle)
+            hFunc(TEvPrivate::TEvDeallocatePersistentBufferChunk, Handle)
 
             hFunc(TEvents::TEvUndelivered, Handle)
 
@@ -292,6 +293,7 @@ namespace {
             hFunc(NPDisk::TEvReadLogResult, Handle)
             cFunc(TEvPrivate::EvHandleSingleQuery, HandleSingleQuery)
             hFunc(NPDisk::TEvChunkReserveResult, Handle)
+            hFunc(NPDisk::TEvChunkForgetResult, Handle)
             hFunc(NPDisk::TEvLogResult, Handle)
             hFunc(TEvPrivate::TEvHandleEventForChunk, Handle)
             hFunc(NPDisk::TEvCutLog, Handle)
@@ -330,6 +332,7 @@ namespace {
             hFunc(TEvents::TEvUndelivered, Handle)
 
             hFunc(TEvPrivate::TEvHandlePersistentBufferEventForChunk, Handle)
+            hFunc(TEvPrivate::TEvDeallocatePersistentBufferChunkResult, Handle)
 
             hFunc(NPDisk::TEvChunkWriteRawResult, Handle)
             hFunc(NPDisk::TEvChunkReadRawResult, Handle)
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor.h b/ydb/core/blobstorage/ddisk/ddisk_actor.h
index 23ad90c8568..8317246120a 100644
--- a/ydb/core/blobstorage/ddisk/ddisk_actor.h
+++ b/ydb/core/blobstorage/ddisk/ddisk_actor.h
@@ -241,11 +241,29 @@ namespace NKikimr::NDDisk {
                 EvReadPersistentBufferPart,
                 EvInternalSyncWriteResult,
                 EvIssuePersistentBufferChunkAllocation,
+                EvDeallocatePersistentBufferChunk,
+                EvDeallocatePersistentBufferChunkResult,
             };
 
             struct TEvIssuePersistentBufferChunkAllocation : TEventLocal<TEvIssuePersistentBufferChunkAllocation, EvIssuePersistentBufferChunkAllocation> {
             };
 
+            struct TEvDeallocatePersistentBufferChunk : TEventLocal<TEvDeallocatePersistentBufferChunk, EvDeallocatePersistentBufferChunk> {
+                ui32 ChunkIdx;
+
+                TEvDeallocatePersistentBufferChunk(ui32 chunkIdx)
+                    : ChunkIdx(chunkIdx)
+                {}
+            };
+
+            struct TEvDeallocatePersistentBufferChunkResult : TEventLocal<TEvDeallocatePersistentBufferChunkResult, EvDeallocatePersistentBufferChunkResult> {
+                ui32 ChunkIdx;
+
+                TEvDeallocatePersistentBufferChunkResult(ui32 chunkIdx)
+                    : ChunkIdx(chunkIdx)
+                {}
+            };
+
             struct TEvHandleEventForChunk : TEventLocal<TEvHandleEventForChunk, EvHandleEventForChunk> {
                 ui64 TabletId;
                 ui64 VChunkIndex;
@@ -333,6 +351,7 @@ namespace NKikimr::NDDisk {
             WakeupUpdateFreeSpaceInfo = 2,
             WakeupCollectPbStats = 3,
             WakeupProcessPersistentBufferBatchWrite = 4,
+            WakeupProcessDeallocatePersistentBufferChunk = 5,
         };
 
         struct TPbOpSnapshot {
@@ -466,6 +485,7 @@ namespace NKikimr::NDDisk {
 
         void IssueChunkAllocation(ui64 tabletId, ui64 vChunkIndex);
         void Handle(NPDisk::TEvChunkReserveResult::TPtr ev);
+        void Handle(NPDisk::TEvChunkForgetResult::TPtr ev);
         void HandleChunkReserved();
         void Handle(NPDisk::TEvLogResult::TPtr ev);
         void Handle(TEvPrivate::TEvHandleEventForChunk::TPtr ev);
@@ -781,6 +801,7 @@ namespace NKikimr::NDDisk {
         void CreatePersistentBuffer();
         void InitPersistentBuffer();
         void IssuePersistentBufferChunkAllocation();
+        void ProcessDeallocatePersistentBufferChunk(bool forceToNextChunk = false);
         void ProcessPersistentBufferQueue();
         std::vector<std::tuple<ui32, ui32, TRope>> SlicePersistentBuffer(ui64 tabletId, ui32 generation, ui64 vchunkIndex, ui64 lsn, ui32 offsetInBytes, ui32 size, TRcBuf&& payloadWithHeader, std::vector<TPersistentBufferSectorInfo>& sectors);
         std::vector<std::tuple<ui32, ui32, TRope>> SlicePersistentBufferData(TRope& data, std::vector<TPersistentBufferSectorInfo>& sectors);
@@ -809,6 +830,8 @@ namespace NKikimr::NDDisk {
         void Handle(TEvents::TEvUndelivered::TPtr ev);
         void Handle(TEvListPersistentBuffer::TPtr ev);
         void Handle(TEvPrivate::TEvIssuePersistentBufferChunkAllocation::TPtr ev);
+        void Handle(TEvPrivate::TEvDeallocatePersistentBufferChunk::TPtr ev);
+        void Handle(TEvPrivate::TEvDeallocatePersistentBufferChunkResult::TPtr ev);
         void Handle(TEvGetPersistentBufferInfo::TPtr ev);
 
         void Handle(TEvWritePersistentBuffers::TPtr ev);
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_chunks.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_chunks.cpp
index 98744994b7e..3b0e083f8b4 100644
--- a/ydb/core/blobstorage/ddisk/ddisk_actor_chunks.cpp
+++ b/ydb/core/blobstorage/ddisk/ddisk_actor_chunks.cpp
@@ -25,6 +25,28 @@ namespace NKikimr::NDDisk {
         }
     }
 
+    void TDDiskActor::Handle(TEvPrivate::TEvDeallocatePersistentBufferChunk::TPtr ev) {
+        auto chunkIdx = ev->Get()->ChunkIdx;
+        auto it = std::find(PersistentBufferChunks.begin(), PersistentBufferChunks.end(), chunkIdx);
+        Y_DEBUG_ABORT_UNLESS(it != PersistentBufferChunks.end());
+        PersistentBufferChunks.erase(it);
+        IssuePDiskLogRecord(TLogSignature::SignaturePersistentBufferChunkMap, chunkIdx
+            , CreatePersistentBufferChunkMapSnapshot(), &PersistentBufferChunkMapSnapshotLsn, [this, chunkIdx] {
+            Send(PersistentBufferActorId, new TEvPrivate::TEvDeallocatePersistentBufferChunkResult(chunkIdx));
+            --*Counters.Chunks.ChunksOwned;
+        });
+        Send(BaseInfo.PDiskActorID, new NPDisk::TEvChunkForget(PDiskParams->Owner, PDiskParams->OwnerRound,
+            {chunkIdx}));
+    }
+
+    void TDDiskActor::Handle(NPDisk::TEvChunkForgetResult::TPtr ev) {
+        auto& msg = *ev->Get();
+        YDB_LOG_DEBUG("TDDiskActor::Handle(TEvChunkForgetResult)",
+            {"marker", "BSDD04"},
+            {"DDiskId", DDiskId},
+            {"msg", msg});
+    }
+
     void TDDiskActor::Handle(NPDisk::TEvChunkReserveResult::TPtr ev) {
         auto& msg = *ev->Get();
         YDB_LOG_DEBUG("TDDiskActor::Handle(TEvChunkReserveResult)",
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_persistent_buffer.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_persistent_buffer.cpp
index 309426f24cd..f06dfe22f56 100644
--- a/ydb/core/blobstorage/ddisk/ddisk_actor_persistent_buffer.cpp
+++ b/ydb/core/blobstorage/ddisk/ddisk_actor_persistent_buffer.cpp
@@ -28,6 +28,47 @@ namespace NKikimr::NDDisk {
                 {"persistentBufferSpaceAllocator", PersistentBufferSpaceAllocator});
         }
     }
+    void TDDiskActor::ProcessDeallocatePersistentBufferChunk(bool forceToNextChunk) {
+        Y_ABORT_UNLESS(IsPersistentBufferActor);
+        ui64 freeSpace = PersistentBufferSpaceAllocator.GetFreeSpace();
+        ui64 ownedChunks = PersistentBufferSpaceAllocator.OwnedChunks.size();
+        bool canDeallocate = freeSpace * 100 > ownedChunks * SectorInChunk * PersistentBufferFormat.DeallocateFreeSpaceThresholdPercent
+            && ownedChunks > PersistentBufferFormat.InitChunks;
+
+        if (PersistentBufferSpaceAllocator.IsChunkLocked() && !canDeallocate) {
+            PersistentBufferSpaceAllocator.UnlockChunk();
+            return;
+        }
+
+        if (canDeallocate) {
+            if (forceToNextChunk) {
+                PersistentBufferSpaceAllocator.UnlockChunk();
+            }
+            if (!PersistentBufferSpaceAllocator.IsChunkLocked()) {
+                PersistentBufferSpaceAllocator.LockNextChunk();
+                Schedule(TDuration::Seconds(PersistentBufferFormat.DeallocateThresholdSeconds), new TEvents::TEvWakeup(EWakeupTag::WakeupProcessDeallocatePersistentBufferChunk));
+            }
+        }
+
+        if (auto deallocateChunkIdx = PersistentBufferSpaceAllocator.LockedChunkIdx; PersistentBufferSpaceAllocator.DeallocateChunk()) {
+            auto ddiskActorId = MakeBlobStorageDDiskId(SelfId().NodeId(), BaseInfo.PDiskId, BaseInfo.VDiskSlotId);
+            Send(ddiskActorId, new TEvPrivate::TEvDeallocatePersistentBufferChunk(deallocateChunkIdx));
+            YDB_LOG_DEBUG_COMP(NKikimrServices::BS_PERSISTENT_BUFFER, "TDDiskActor::ProcessDeallocatePersistentBufferChunk deallocate chunk",
+                {"marker", "BSPB"},
+                {"PBufferId", SelfId()},
+                {"freeSpace", PersistentBufferSpaceAllocator.GetFreeSpace()},
+                {"deallocateChunkIdx", deallocateChunkIdx});
+        }
+    }
+
+    void TDDiskActor::Handle(TEvPrivate::TEvDeallocatePersistentBufferChunkResult::TPtr ev) {
+        auto& msg = *ev->Get();
+        YDB_LOG_DEBUG_COMP(NKikimrServices::BS_PERSISTENT_BUFFER, "TDDiskActor::TEvDeallocatePersistentBufferChunkResult",
+            {"marker", "BSPB"},
+            {"PBufferId", SelfId()},
+            {"msg", msg});
+
+    }
 
     void TDDiskActor::InitPersistentBuffer() {
         Y_ABORT_UNLESS(IsPersistentBufferActor);
@@ -1655,6 +1696,7 @@ namespace NKikimr::NDDisk {
             } else {
                 PersistentBufferSpaceAllocator.Free({pr.Sectors.begin() + 1, pr.Sectors.end()});
             }
+            ProcessDeallocatePersistentBufferChunk();
 
             buffer.Size -= pr.Size;
             for (auto readCookie : pr.ReadInflight) {
diff --git a/ydb/core/blobstorage/ddisk/ddisk_actor_read_write.cpp b/ydb/core/blobstorage/ddisk/ddisk_actor_read_write.cpp
index 5c5e0d3bffe..37128155414 100644
--- a/ydb/core/blobstorage/ddisk/ddisk_actor_read_write.cpp
+++ b/ydb/core/blobstorage/ddisk/ddisk_actor_read_write.cpp
@@ -390,6 +390,10 @@ namespace NKikimr::NDDisk {
                 ProcessPersistentBufferBatchWrite();
                 break;
             }
+            case EWakeupTag::WakeupProcessDeallocatePersistentBufferChunk: {
+                ProcessDeallocatePersistentBufferChunk(true);
+                break;
+            }
         }
     }
 
diff --git a/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.cpp b/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.cpp
index 4a29a30a168..4f603781a0e 100644
--- a/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.cpp
+++ b/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.cpp
@@ -219,9 +219,19 @@ namespace NKikimr::NDDisk {
         if (FreeSpace < sectorsCount) {
             return result;
         }
+        if (FreeSpace < SectorsInChunk + sectorsCount) {
+            auto it = FreeSpaceMap.find(LockedChunkIdx);
+            if (it != FreeSpaceMap.end() && FreeSpace - it->second.FreeSpace < sectorsCount) {
+                UnlockChunk();
+            }
+        }
         while (result.size() != sectorsCount) {
             Y_ABORT_UNLESS(!ChunksByFreeSpace.empty());
             ui32 chunkIdx = ChunksByFreeSpace.back().ChunkIdx;
+            if (chunkIdx == LockedChunkIdx) {
+                Y_ABORT_UNLESS(ChunksByFreeSpace.size() >= 2);
+                chunkIdx = ChunksByFreeSpace[ChunksByFreeSpace.size() - 2].ChunkIdx;
+            }
             auto it = FreeSpaceMap.find(chunkIdx);
             Y_ABORT_UNLESS(it != FreeSpaceMap.end());
             it->second.Occupy(sectorsCount, result);
@@ -320,5 +330,42 @@ namespace NKikimr::NDDisk {
         return fs;
     }
 
+    void TPersistentBufferSpaceAllocator::LockNextChunk() {
+        if (LockedChunkIdx != Max<ui32>()) {
+            return;
+        }
+        Y_ABORT_UNLESS(OwnedChunks.size() > 0);
+        LastLockedOwnedChunksIdx++;
+        if (LastLockedOwnedChunksIdx >= OwnedChunks.size()) {
+            LastLockedOwnedChunksIdx = 0;
+        }
+        LockedChunkIdx = OwnedChunks[LastLockedOwnedChunksIdx];
+    }
+
+    void TPersistentBufferSpaceAllocator::UnlockChunk() {
+        LockedChunkIdx = Max<ui32>();
+    }
+
+    bool TPersistentBufferSpaceAllocator::DeallocateChunk() {
+        auto it = FreeSpaceMap.find(LockedChunkIdx);
+        if (it == FreeSpaceMap.end() || it->second.FreeSpace != SectorsInChunk) {
+            return false;
+        }
+        FreeSpaceMap.erase(LockedChunkIdx);
+        // A fully-free chunk (FreeSpace == SectorsInChunk) has the maximum possible free space,
+        // so in ChunksByFreeSpace (sorted ascending by FreeSpace) it is located at or near the
+        // very end of the vector. Locate it with a binary search on the unique {ChunkIdx, FreeSpace}
+        // key instead of scanning the whole vector, and erase it directly.
+        auto chunkIter = std::lower_bound(ChunksByFreeSpace.begin(), ChunksByFreeSpace.end(),
+            TChunkInfo{LockedChunkIdx, SectorsInChunk}, ChunksByFreeSpaceLess);
+        Y_ABORT_UNLESS(chunkIter != ChunksByFreeSpace.end()
+            && chunkIter->ChunkIdx == LockedChunkIdx && chunkIter->FreeSpace == SectorsInChunk);
+        ChunksByFreeSpace.erase(chunkIter);
+        std::erase(OwnedChunks, LockedChunkIdx);
+        FreeSpace -= SectorsInChunk;
+        Y_DEBUG_ABORT_UNLESS(FreeSpace == VerifyFreeSpace());
+        UnlockChunk();
+        return true;
+    }
 
 } // NKikimr::NDDisk
diff --git a/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.h b/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.h
index 52dc93a19f2..a5876b7c9b1 100644
--- a/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.h
+++ b/ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.h
@@ -65,6 +65,9 @@ namespace NKikimr::NDDisk {
         ui32 VerifyFreeSpace();
 
     public:
+        ui32 LastLockedOwnedChunksIdx = Max<ui32>();
+        ui32 LockedChunkIdx = Max<ui32>();
+
         std::vector<ui32> OwnedChunks;
 
         TPersistentBufferSpaceAllocator(ui32 sectorsInChunk = 32768);
@@ -73,6 +76,12 @@ namespace NKikimr::NDDisk {
         void Free(const std::span<TPersistentBufferSectorInfo> locations);
         void MarkOccupied(std::span<const TPersistentBufferSectorInfo> locations);
         void AddNewChunk(ui32 chunkIdx);
+        void LockNextChunk();
+        bool IsChunkLocked() const {
+            return LockedChunkIdx != Max<ui32>();
+        }
+        void UnlockChunk();
+        bool DeallocateChunk();
         ui32 GetFreeSpace() const {
             return FreeSpace;
         }
diff --git a/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp b/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp
index 55487cc1d79..5aa8cd65882 100644
--- a/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp
+++ b/ydb/core/blobstorage/ddisk/ut/ddisk_actor_ut.cpp
@@ -3281,6 +3281,214 @@ Y_UNIT_TEST_SUITE(TDDiskActorTest) {
         UNIT_ASSERT_VALUES_EQUAL(GetPBAllocatedChunks(ctx, disk), PersistentBufferInitChunks);
     }
 
+    // ─────────────────────────────────────────────────────────────────────────
+    // Integration tests for proactive chunk deallocation
+    // (TPersistentBufferFormat::DeallocateFreeSpaceThresholdPercent /
+    //  DeallocateThresholdSeconds).
+    //
+    // ProcessDeallocatePersistentBufferChunk (called after every Free()) locks
+    // owned chunks round-robin and, once a locked chunk turns out to be fully
+    // free, sends TEvPrivate::TEvDeallocatePersistentBufferChunk to the DDisk
+    // actor, which writes a chunk-map log record and issues NPDisk::TEvChunkForget
+    // for the physical chunk.
+    //
+    // Shared setup: reuse MakeProactiveAllocationFormat / DoPBWriteRoundTrip /
+    // GetPBFreeSectors from the proactive-allocation tests above. 12 writes with
+    // PreallocateFreeSpaceThresholdPercent = 99 leave the buffer with 5 owned
+    // chunks (4 original + 1 proactively allocated), where the 5th chunk (index
+    // PersistentBufferInitChunks) is fully free -- exactly the state needed to
+    // exercise deallocation.
+    // ─────────────────────────────────────────────────────────────────────────
+
+    NDDisk::TPersistentBufferFormat MakeDeallocationFormat(ui32 deallocateFreeSpaceThresholdPercent, ui32 deallocateThresholdSeconds) {
+        NDDisk::TPersistentBufferFormat fmt = MakeProactiveAllocationFormat(256 /*maxChunks*/, 99 /*preallocateFreeSpaceThresholdPercent*/);
+        fmt.DeallocateFreeSpaceThresholdPercent = deallocateFreeSpaceThresholdPercent;
+        fmt.DeallocateThresholdSeconds = deallocateThresholdSeconds;
+        return fmt;
+    }
+
+    // Drives 12 writes (as in PersistentBufferProactiveChunkAllocation) to reach
+    // PersistentBufferInitChunks + 1 owned chunks, with the extra (last) chunk
+    // fully free. Returns the physical chunk id of that extra chunk.
+    ui32 ReachFiveChunksWithLastFullyFree(TTestContext& ctx, const TDiskHandle& disk, const NDDisk::TQueryCredentials& creds) {
+        for (ui32 i = 0; i < 11; ++i) {
+            DoPBWriteRoundTrip(ctx, disk, creds, /*lsn=*/10 + i, 'A' + i);
+        }
+        {
+            const ui32 writeSize = BlockSize * 128;
+            const NDDisk::TBlockSelector selector{3, 0, writeSize};
+            auto write = std::make_unique<NDDisk::TEvWritePersistentBuffer>(
+                creds, selector, /*lsn=*/21, NDDisk::TWriteInstruction(0));
+            write->AddPayload(TRope(MakeData('M', writeSize)));
+            SendToDDisk(ctx, disk.PBServiceId, write.release());
+
+            auto pbWriteRaw = ctx.WaitPDiskRequest<NPDisk::TEvChunkWriteRaw>(disk);
+            ctx.SendPDiskResponse(disk, *pbWriteRaw, new NPDisk::TEvChunkWriteRawResult(NKikimrProto::OK, ""));
+
+            auto log = ctx.WaitPDiskRequest<NPDisk::TEvLog>(disk);
+            auto logReply = std::make_unique<NPDisk::TEvLogResult>(NKikimrProto::OK, 0, "", 0);
+            logReply->Results.emplace_back(log->Get()->Lsn, log->Get()->Cookie);
+            ctx.SendPDiskResponse(disk, *log, logReply.release());
+
+            auto reserve = ctx.WaitPDiskRequest<NPDisk::TEvChunkReserve>(disk);
+            UNIT_ASSERT_VALUES_EQUAL(reserve->Get()->SizeChunks, 1u);
+            auto reserveReply = std::make_unique<NPDisk::TEvChunkReserveResult>(NKikimrProto::OK, 0);
+            reserveReply->ChunkIds.push_back(disk.FirstChunkId + PersistentBufferInitChunks + MinChunksReserved);
+            ctx.SendPDiskResponse(disk, *reserve, reserveReply.release());
+
+            auto writeResult = WaitFromDDisk<NDDisk::TEvWritePersistentBufferResult>(ctx);
+            AssertStatus(writeResult, TReplyStatus::OK);
+        }
+        UNIT_ASSERT_VALUES_EQUAL(GetPBAllocatedChunks(ctx, disk), PersistentBufferInitChunks + 1);
+        return disk.FirstChunkId + PersistentBufferInitChunks; // the proactively allocated (5th) chunk
+    }
+
+    // Erase lsn=10 (the very first write, on the very first owned chunk) via
+    // TEvBatchErasePersistentBuffer with fast erases effectively bypassed
+    // (EnableFastErases = false in the format), so the erase goes through the
+    // plain ErasePersistentBuffer -> ClearPersistentBufferRecords path, which is
+    // the one that calls ProcessDeallocatePersistentBufferChunk().
+    void EraseFirstRecordSlowPath(TTestContext& ctx, const TDiskHandle& disk, const NDDisk::TQueryCredentials& creds) {
+        auto batchErase = std::make_unique<NDDisk::TEvBatchErasePersistentBuffer>(creds);
+        batchErase->AddErase(/*lsn=*/10, creds.Generation);
+        SendToDDisk(ctx, disk.PBServiceId, batchErase.release());
+
+        auto eraseRaw = ctx.WaitPDiskRequest<NPDisk::TEvChunkWriteRaw>(disk);
+        ctx.SendPDiskResponse(disk, *eraseRaw, new NPDisk::TEvChunkWriteRawResult(NKikimrProto::OK, ""));
+
+        auto eraseResult = WaitFromDDisk<NDDisk::TEvErasePersistentBufferResult>(ctx);
+        AssertStatus(eraseResult, TReplyStatus::OK);
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // Test: after 12 writes leave a 5th, fully-free chunk and DeallocateFreeSpaceThresholdPercent
+    // is set very low (so the free-space precondition is always true once the buffer can shrink),
+    // erasing a record frees sectors and triggers ProcessDeallocatePersistentBufferChunk, which
+    // round-robins the lock through the owned chunks (starting at chunk 0) until it reaches the
+    // fully-free 5th chunk, then issues TEvPrivate::TEvDeallocatePersistentBufferChunk -> a
+    // persistent-buffer-chunk-map log record -> NPDisk::TEvChunkForget for that physical chunk.
+    //
+    // Covers: ProcessDeallocatePersistentBufferChunk, TPersistentBufferSpaceAllocator::LockNextChunk/
+    //         DeallocateChunk, TDDiskActor::Handle(TEvDeallocatePersistentBufferChunk),
+    //         TDDiskActor::Handle(TEvChunkForgetResult).
+    // ─────────────────────────────────────────────────────────────────────────
+    Y_UNIT_TEST(PersistentBufferProactiveDeallocationAfterErase) {
+        TTestContext ctx;
+        auto fmt = MakeDeallocationFormat(/*deallocateFreeSpaceThresholdPercent=*/90, /*deallocateThresholdSeconds=*/1);
+        fmt.EnableFastErases = false;
+        const TDiskHandle disk = ctx.CreateDDisk(40, 1, fmt);
+        NDDisk::TQueryCredentials creds = Connect(ctx, disk.PBServiceId, 90, 1);
+
+        const ui32 extraChunk = ReachFiveChunksWithLastFullyFree(ctx, disk, creds);
+        const ui32 freeSectorsBeforeErase = GetPBFreeSectors(ctx, disk);
+
+        // Erase the very first write (lsn=10, physically on chunk #0): frees 129
+        // sectors and triggers ProcessDeallocatePersistentBufferChunk(). The
+        // free-space precondition is satisfied (5 owned chunks, well above the
+        // 90% threshold), so the allocator starts round-robin locking, beginning
+        // at chunk #0. Chunks 0..3 all still hold occupied sectors from the 12
+        // writes, so each lock attempt fails and reschedules a 1-second wakeup
+        // with forceToNextChunk=true, advancing the lock to the next chunk. Only
+        // the 5th chunk (never written to) is fully free, so the deallocation
+        // succeeds on the 5th lock attempt (chunks 0,1,2,3,4).
+        EraseFirstRecordSlowPath(ctx, disk, creds);
+
+        UNIT_ASSERT_VALUES_EQUAL(GetPBFreeSectors(ctx, disk), freeSectorsBeforeErase + 129);
+
+        // Drain the persistent-buffer-chunk-map log record for the deallocation
+        // (the simulated clock advances through the intermediate 1-second wakeup
+        // cycles automatically while waiting for this event).
+        auto log = ctx.WaitPDiskRequest<NPDisk::TEvLog>(disk);
+        auto logReply = std::make_unique<NPDisk::TEvLogResult>(NKikimrProto::OK, 0, "", 0);
+        logReply->Results.emplace_back(log->Get()->Lsn, log->Get()->Cookie);
+        ctx.SendPDiskResponse(disk, *log, logReply.release());
+
+        // The DDisk actor must now ask PDisk to forget exactly the extra chunk.
+        auto forget = ctx.WaitPDiskRequest<NPDisk::TEvChunkForget>(disk);
+        UNIT_ASSERT_VALUES_EQUAL(forget->Get()->ForgetChunks.size(), 1u);
+        UNIT_ASSERT_VALUES_EQUAL(forget->Get()->ForgetChunks[0], extraChunk);
+        ctx.SendPDiskResponse(disk, *forget, new NPDisk::TEvChunkForgetResult(NKikimrProto::OK, 0));
+
+        // The deallocated chunk's capacity (32768 sectors) must be gone from the
+        // free pool: free sectors after deallocation must be exactly
+        // (freeSectorsBeforeErase + 129 [reclaimed by the erase] - 32768 [chunk
+        // capacity removed]).
+        constexpr ui32 SectorsPerChunk = TTestContext::ChunkSize / BlockSize;
+        UNIT_ASSERT_VALUES_EQUAL(GetPBFreeSectors(ctx, disk),
+            freeSectorsBeforeErase + 129 - SectorsPerChunk);
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // Test: DeallocateFreeSpaceThresholdPercent = 100 disables proactive
+    // deallocation completely (freeSpace * 100 > ownedChunks * SectorInChunk * 100
+    // is never true), even though a 5th, fully-free chunk exists and an erase
+    // frees additional sectors.
+    //
+    // Covers: ProcessDeallocatePersistentBufferChunk -> canDeallocate threshold
+    //         sub-condition.
+    // ─────────────────────────────────────────────────────────────────────────
+    Y_UNIT_TEST(PersistentBufferDeallocationDisabledByFullThreshold) {
+        TTestContext ctx;
+        auto fmt = MakeDeallocationFormat(/*deallocateFreeSpaceThresholdPercent=*/100, /*deallocateThresholdSeconds=*/1);
+        fmt.EnableFastErases = false;
+        const TDiskHandle disk = ctx.CreateDDisk(41, 1, fmt);
+        NDDisk::TQueryCredentials creds = Connect(ctx, disk.PBServiceId, 91, 1);
+
+        ReachFiveChunksWithLastFullyFree(ctx, disk, creds);
+        const ui32 freeSectorsBeforeErase = GetPBFreeSectors(ctx, disk);
+
+        EraseFirstRecordSlowPath(ctx, disk, creds);
+        UNIT_ASSERT_VALUES_EQUAL(GetPBFreeSectors(ctx, disk), freeSectorsBeforeErase + 129);
+
+        // No deallocation may be issued: verify no TEvLog / TEvChunkForget shows
+        // up at the PDisk edge by racing a sentinel wakeup through the same
+        // edge actor. If a chunk-map log request were in flight it would arrive
+        // before the sentinel (FIFO per-actor delivery), causing the assertion
+        // below to observe the log/forget event's recipient instead of the
+        // sentinel edge.
+        TActorId sentinelEdge = ctx.Runtime.AllocateEdgeActor(NodeId, __FILE__, __LINE__);
+        ctx.Runtime.Send(new IEventHandle(sentinelEdge, ctx.Edge, new TEvents::TEvWakeup()), NodeId);
+        auto ev = ctx.Runtime.WaitForEdgeActorEvent({disk.PDiskEdge, sentinelEdge});
+        UNIT_ASSERT_VALUES_EQUAL_C(ev->Recipient, sentinelEdge,
+            "no PDisk request (deallocation) should be issued when DeallocateFreeSpaceThresholdPercent=100");
+
+        // Still exactly PersistentBufferInitChunks + 1 owned chunks worth of free space.
+        UNIT_ASSERT_VALUES_EQUAL(GetPBFreeSectors(ctx, disk), freeSectorsBeforeErase + 129);
+    }
+
+    // ─────────────────────────────────────────────────────────────────────────
+    // Test: deallocation must not fire while the buffer owns exactly InitChunks
+    // chunks, even if free space is at 100% (canDeallocate requires
+    // ownedChunks > InitChunks).
+    //
+    // Covers: ProcessDeallocatePersistentBufferChunk -> "ownedChunks > InitChunks"
+    //         sub-condition of canDeallocate.
+    // ─────────────────────────────────────────────────────────────────────────
+    Y_UNIT_TEST(PersistentBufferDeallocationSkippedAtInitChunks) {
+        TTestContext ctx;
+        auto fmt = MakeDeallocationFormat(/*deallocateFreeSpaceThresholdPercent=*/1, /*deallocateThresholdSeconds=*/1);
+        fmt.EnableFastErases = false;
+        fmt.PreallocateFreeSpaceThresholdPercent = 0; // keep exactly InitChunks owned chunks
+        const TDiskHandle disk = ctx.CreateDDisk(42, 1, fmt);
+        NDDisk::TQueryCredentials creds = Connect(ctx, disk.PBServiceId, 92, 1);
+
+        DoPBWriteRoundTrip(ctx, disk, creds, /*lsn=*/10, 'A');
+        UNIT_ASSERT_VALUES_EQUAL(GetPBAllocatedChunks(ctx, disk), PersistentBufferInitChunks);
+
+        EraseFirstRecordSlowPath(ctx, disk, creds);
+
+        // No deallocation may be issued: the buffer owns exactly InitChunks
+        // chunks, so canDeallocate's "ownedChunks > InitChunks" sub-condition is
+        // always false, regardless of how low the free-space threshold is set.
+        TActorId sentinelEdge = ctx.Runtime.AllocateEdgeActor(NodeId, __FILE__, __LINE__);
+        ctx.Runtime.Send(new IEventHandle(sentinelEdge, ctx.Edge, new TEvents::TEvWakeup()), NodeId);
+        auto ev = ctx.Runtime.WaitForEdgeActorEvent({disk.PDiskEdge, sentinelEdge});
+        UNIT_ASSERT_VALUES_EQUAL_C(ev->Recipient, sentinelEdge,
+            "no PDisk request (deallocation) should be issued when ownedChunks == InitChunks");
+
+        UNIT_ASSERT_VALUES_EQUAL(GetPBAllocatedChunks(ctx, disk), PersistentBufferInitChunks);
+    }
+
 } // Y_UNIT_TEST_SUITE
 
 } // NKikimr
diff --git a/ydb/core/blobstorage/ddisk/ut/persistent_buffer_space_allocator_ut.cpp b/ydb/core/blobstorage/ddisk/ut/persistent_buffer_space_allocator_ut.cpp
new file mode 100644
index 00000000000..bac535c4450
--- /dev/null
+++ b/ydb/core/blobstorage/ddisk/ut/persistent_buffer_space_allocator_ut.cpp
@@ -0,0 +1,169 @@
+#include <ydb/core/blobstorage/ddisk/persistent_buffer_space_allocator.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+namespace NKikimr::NDDisk {
+
+namespace {
+
+TPersistentBufferSpaceAllocator MakeAllocator(ui32 sectorsInChunk = 16) {
+    return TPersistentBufferSpaceAllocator(sectorsInChunk);
+}
+
+} // namespace
+
+Y_UNIT_TEST_SUITE(TPersistentBufferSpaceAllocatorLockDeallocateTest) {
+
+    Y_UNIT_TEST(LockNextChunkPicksOwnedChunk) {
+        auto alloc = MakeAllocator();
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+        alloc.AddNewChunk(2);
+
+        UNIT_ASSERT(!alloc.IsChunkLocked());
+        alloc.LockNextChunk();
+        UNIT_ASSERT(alloc.IsChunkLocked());
+        UNIT_ASSERT(std::find(alloc.OwnedChunks.begin(), alloc.OwnedChunks.end(), alloc.LockedChunkIdx)
+            != alloc.OwnedChunks.end());
+    }
+
+    Y_UNIT_TEST(LockNextChunkIsNoopWhenAlreadyLocked) {
+        auto alloc = MakeAllocator();
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+
+        alloc.LockNextChunk();
+        ui32 locked = alloc.LockedChunkIdx;
+        alloc.LockNextChunk();
+        UNIT_ASSERT_VALUES_EQUAL(locked, alloc.LockedChunkIdx);
+    }
+
+    Y_UNIT_TEST(LockNextChunkRoundRobinsThroughOwnedChunks) {
+        auto alloc = MakeAllocator();
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+        alloc.AddNewChunk(2);
+
+        std::vector<ui32> lockedOrder;
+        for (ui32 i = 0; i < 3; ++i) {
+            alloc.LockNextChunk();
+            lockedOrder.push_back(alloc.LockedChunkIdx);
+            alloc.UnlockChunk();
+        }
+        // The three chunks should each be visited exactly once as we round-robin.
+        std::vector<ui32> sorted = lockedOrder;
+        std::sort(sorted.begin(), sorted.end());
+        UNIT_ASSERT_VALUES_EQUAL(sorted.size(), 3u);
+        UNIT_ASSERT_VALUES_EQUAL(sorted[0], 0u);
+        UNIT_ASSERT_VALUES_EQUAL(sorted[1], 1u);
+        UNIT_ASSERT_VALUES_EQUAL(sorted[2], 2u);
+    }
+
+    Y_UNIT_TEST(UnlockChunkResetsLock) {
+        auto alloc = MakeAllocator();
+        alloc.AddNewChunk(0);
+
+        alloc.LockNextChunk();
+        UNIT_ASSERT(alloc.IsChunkLocked());
+        alloc.UnlockChunk();
+        UNIT_ASSERT(!alloc.IsChunkLocked());
+        UNIT_ASSERT_VALUES_EQUAL(alloc.LockedChunkIdx, Max<ui32>());
+    }
+
+    Y_UNIT_TEST(DeallocateChunkFailsWhenNothingLocked) {
+        auto alloc = MakeAllocator();
+        alloc.AddNewChunk(0);
+
+        UNIT_ASSERT(!alloc.DeallocateChunk());
+        UNIT_ASSERT_VALUES_EQUAL(alloc.OwnedChunks.size(), 1u);
+    }
+
+    Y_UNIT_TEST(DeallocateChunkFailsWhenLockedChunkHasOccupiedSectors) {
+        auto alloc = MakeAllocator(16);
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+
+        // Occupy a sector so chunk 0 is not fully free (chunk with most free space is picked last
+        // by Occupy(), i.e. chunk 1; force occupying explicitly on chunk 0 by depleting chunk 1 first).
+        auto sectors = alloc.Occupy(16); // fully occupies one whole chunk (the one with most free space)
+        UNIT_ASSERT_VALUES_EQUAL(sectors.size(), 16u);
+        ui32 occupiedChunk = sectors.front().ChunkIdx;
+
+        alloc.LockedChunkIdx = occupiedChunk;
+        UNIT_ASSERT(!alloc.DeallocateChunk());
+        UNIT_ASSERT_VALUES_EQUAL(alloc.OwnedChunks.size(), 2u);
+    }
+
+    Y_UNIT_TEST(DeallocateChunkSucceedsWhenLockedChunkFullyFree) {
+        auto alloc = MakeAllocator(16);
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+
+        alloc.LockedChunkIdx = 1;
+        UNIT_ASSERT(alloc.DeallocateChunk());
+        UNIT_ASSERT_VALUES_EQUAL(alloc.OwnedChunks.size(), 1u);
+        UNIT_ASSERT_VALUES_EQUAL(alloc.OwnedChunks[0], 0u);
+        UNIT_ASSERT_VALUES_EQUAL(alloc.GetFreeSpace(), 16u);
+        // DeallocateChunk should unlock automatically on success.
+        UNIT_ASSERT(!alloc.IsChunkLocked());
+    }
+
+    Y_UNIT_TEST(DeallocateChunkRemovesChunkFromFutureOccupy) {
+        auto alloc = MakeAllocator(4);
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+
+        alloc.LockedChunkIdx = 1;
+        UNIT_ASSERT(alloc.DeallocateChunk());
+
+        // Only chunk 0 should remain; occupying more than its capacity must fail.
+        auto sectors = alloc.Occupy(5);
+        UNIT_ASSERT_VALUES_EQUAL(sectors.size(), 0u);
+
+        auto sectors2 = alloc.Occupy(4);
+        UNIT_ASSERT_VALUES_EQUAL(sectors2.size(), 4u);
+        for (auto& s : sectors2) {
+            UNIT_ASSERT_VALUES_EQUAL(static_cast<ui32>(s.ChunkIdx), 0u);
+        }
+    }
+
+    Y_UNIT_TEST(OccupySkipsLockedChunkWhenOtherSpaceAvailable) {
+        auto alloc = MakeAllocator(4);
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+
+        // Lock chunk 1 (fully free) to protect it from being used for allocation.
+        alloc.LockedChunkIdx = 1;
+
+        // Requesting all 4 sectors of a single chunk should come from chunk 0 (unlocked),
+        // leaving chunk 1 fully free and lockable/deallocatable.
+        auto sectors = alloc.Occupy(4);
+        UNIT_ASSERT_VALUES_EQUAL(sectors.size(), 4u);
+        for (auto& s : sectors) {
+            UNIT_ASSERT_VALUES_EQUAL(static_cast<ui32>(s.ChunkIdx), 0u);
+        }
+        UNIT_ASSERT(alloc.DeallocateChunk());
+    }
+
+    Y_UNIT_TEST(OccupyUnlocksChunkWhenNoOtherSpaceAvailable) {
+        auto alloc = MakeAllocator(4);
+        alloc.AddNewChunk(0);
+        alloc.AddNewChunk(1);
+
+        // Occupy all of chunk 0 so it's no longer free.
+        auto used = alloc.Occupy(4);
+        UNIT_ASSERT_VALUES_EQUAL(used.size(), 4u);
+
+        // Now lock the only remaining fully-free chunk (chunk 1).
+        alloc.LockedChunkIdx = 1;
+
+        // Free space is only within the locked chunk; Occupy must fall back to using it
+        // (unlocking automatically) rather than failing.
+        auto sectors = alloc.Occupy(2);
+        UNIT_ASSERT_VALUES_EQUAL(sectors.size(), 2u);
+        UNIT_ASSERT(!alloc.IsChunkLocked());
+    }
+
+} // Y_UNIT_TEST_SUITE(TPersistentBufferSpaceAllocatorLockDeallocateTest)
+
+} // namespace NKikimr::NDDisk
diff --git a/ydb/core/blobstorage/ddisk/ut/ya.make b/ydb/core/blobstorage/ddisk/ut/ya.make
index 70ef76463e3..c425ca80acd 100644
--- a/ydb/core/blobstorage/ddisk/ut/ya.make
+++ b/ydb/core/blobstorage/ddisk/ut/ya.make
@@ -18,6 +18,7 @@ SRCS(
     ddisk_actor_pdisk_ut.cpp
     ddisk_sync_ut.cpp
     persistent_buffer_barriers_manager_ut.cpp
+    persistent_buffer_space_allocator_ut.cpp
     segment_manager_ut.cpp
 )
 
diff --git a/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp b/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
index d9cdbf01d06..556df00aa88 100644
--- a/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
+++ b/ydb/core/blobstorage/nodewarden/node_warden_vdisk.cpp
@@ -269,6 +269,26 @@ namespace NKikimr::NStorage {
                         pbufferFormat.PreallocateFreeSpaceThresholdPercent = newValue;
                     }
                 }
+                if (Cfg->PBufferConfig->HasDeallocateFreeSpaceThresholdPercent()) {
+                    auto newValue = Cfg->PBufferConfig->GetDeallocateFreeSpaceThresholdPercent();
+                    if (newValue > 100) {
+                        YDB_LOG_ERROR("DeallocateFreeSpaceThresholdPercent value should be less or equal to 100",
+                            {"marker", "NW37"},
+                            {"DeallocateFreeSpaceThresholdPercent", newValue});
+                    } else {
+                        pbufferFormat.DeallocateFreeSpaceThresholdPercent = newValue;
+                    }
+                }
+                if (Cfg->PBufferConfig->HasDeallocateThresholdSeconds()) {
+                    auto newValue = Cfg->PBufferConfig->GetDeallocateThresholdSeconds();
+                    if (newValue > 600) {
+                        YDB_LOG_ERROR("DeallocateThresholdSeconds value should be less or equal to 600",
+                            {"marker", "NW38"},
+                            {"DeallocateThresholdSeconds", newValue});
+                    } else {
+                        pbufferFormat.DeallocateThresholdSeconds = newValue;
+                    }
+                }
             }
             actor.reset(NDDisk::CreateDDiskActor(std::move(baseInfo), groupInfo, std::move(pbufferFormat),
                 std::move(ddiskConfig), AppData()->Counters));
diff --git a/ydb/core/nbs/cloud/blockstore/config/protos/ddisk_config.proto b/ydb/core/nbs/cloud/blockstore/config/protos/ddisk_config.proto
index 2087ca11073..2649ae72ed8 100644
--- a/ydb/core/nbs/cloud/blockstore/config/protos/ddisk_config.proto
+++ b/ydb/core/nbs/cloud/blockstore/config/protos/ddisk_config.proto
@@ -53,4 +53,12 @@ message TPBufferConfig
     // Allocate a new chunk proactively when free space drops below this percentage
     // of the currently owned capacity. 0 disables proactive allocation. Default 10%.
     optional uint32 PreallocateFreeSpaceThresholdPercent = 13;
+
+    // Deallocate a chunk proactively when free space is over this percentage
+    // of the currently owned capacity. 100% disables proactive deallocation. Default 90%.
+    optional uint32 DeallocateFreeSpaceThresholdPercent = 14;
+
+    // Deallocate a chunk proactively when it has been freed for this many seconds.
+    // Default 30 seconds.
+    optional uint32 DeallocateThresholdSeconds = 15;
 }