Make BlobDepot cut tablet channel history whenever possible KIKIMR-18366

4 files changed, 75 insertions, 18 deletions

diff --git a/ydb/core/blob_depot/data.h b/ydb/core/blob_depot/data.h
index f66fb684b6..4f77b28391 100644
--- a/ydb/core/blob_depot/data.h
+++ b/ydb/core/blob_depot/data.h
@@ -417,6 +417,7 @@ namespace NKikimr::NBlobDepot {
         THashMap<TLogoBlobID, ui32> RefCount;
         THashMap<std::tuple<ui8, ui32>, TRecordsPerChannelGroup> RecordsPerChannelGroup;
         std::optional<TLogoBlobID> LastAssimilatedBlobId;
+        THashSet<std::tuple<ui8, ui32>> AlreadyCutHistory;
         ui64 TotalStoredDataSize = 0;
         ui64 TotalStoredTrashSize = 0;
         ui64 InFlightTrashSize = 0;
@@ -643,7 +644,7 @@ namespace NKikimr::NBlobDepot {
         void HandleTrash(TRecordsPerChannelGroup& record);
         void Handle(TEvBlobStorage::TEvCollectGarbageResult::TPtr ev);
         void OnPushNotifyResult(TEvBlobDepot::TEvPushNotifyResult::TPtr ev);
-        void OnCommitConfirmedGC(ui8 channel, ui32 groupId);
+        void OnCommitConfirmedGC(ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted);
         bool OnBarrierShift(ui64 tabletId, ui8 channel, bool hard, TGenStep previous, TGenStep current, ui32& maxItems,
             NTabletFlatExecutor::TTransactionContext& txc, void *cookie);
 
@@ -716,7 +717,8 @@ namespace NKikimr::NBlobDepot {
         void ExecuteIssueGC(ui8 channel, ui32 groupId, TGenStep issuedGenStep,
             std::unique_ptr<TEvBlobStorage::TEvCollectGarbage> collectGarbage, ui64 cookie);
 
-        void ExecuteConfirmGC(ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted, TGenStep confirmedGenStep);
+        void ExecuteConfirmGC(ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted, size_t index,
+            TGenStep confirmedGenStep);
     };
 
     Y_DECLARE_OPERATORS_FOR_FLAGS(TBlobDepot::TData::TScanFlags);
diff --git a/ydb/core/blob_depot/data_gc.cpp b/ydb/core/blob_depot/data_gc.cpp
index 5fdacbd39f..d4dc4251eb 100644
--- a/ydb/core/blob_depot/data_gc.cpp
+++ b/ydb/core/blob_depot/data_gc.cpp
@@ -45,6 +45,7 @@ namespace NKikimr::NBlobDepot {
         const ui8 Channel;
         const ui32 GroupId;
         std::vector<TLogoBlobID> TrashDeleted;
+        size_t Index;
         const TGenStep ConfirmedGenStep;
 
         static constexpr ui32 MaxKeysToProcessAtOnce = 10'000;
@@ -52,43 +53,42 @@ namespace NKikimr::NBlobDepot {
     public:
         TTxType GetTxType() const override { return NKikimrBlobDepot::TXTYPE_CONFIRM_GC; }
 
-        TTxConfirmGC(TBlobDepot *self, ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted, TGenStep confirmedGenStep)
+        TTxConfirmGC(TBlobDepot *self, ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted, size_t index,
+                TGenStep confirmedGenStep)
             : TTransactionBase(self)
             , Channel(channel)
             , GroupId(groupId)
             , TrashDeleted(std::move(trashDeleted))
+            , Index(index)
             , ConfirmedGenStep(confirmedGenStep)
         {}
 
         bool Execute(TTransactionContext& txc, const TActorContext&) override {
             NIceDb::TNiceDb db(txc.DB);
 
-            for (ui32 i = 0; i < TrashDeleted.size() && i < MaxKeysToProcessAtOnce; ++i) {
-                db.Table<Schema::Trash>().Key(TKey(TrashDeleted[i]).MakeBinaryKey()).Delete();
+            for (ui32 i = 0; Index < TrashDeleted.size() && i < MaxKeysToProcessAtOnce; ++i, ++Index) {
+                db.Table<Schema::Trash>().Key(TKey(TrashDeleted[Index]).MakeBinaryKey()).Delete();
             }
-            if (TrashDeleted.size() <= MaxKeysToProcessAtOnce) {
-                TrashDeleted.clear();
+            if (Index == TrashDeleted.size()) {
                 db.Table<Schema::GC>().Key(Channel, GroupId).Update<Schema::GC::ConfirmedGenStep>(ui64(ConfirmedGenStep));
-            } else {
-                std::vector<TLogoBlobID> temp;
-                temp.insert(temp.end(), TrashDeleted.begin() + MaxKeysToProcessAtOnce, TrashDeleted.end());
-                temp.swap(TrashDeleted);
             }
 
             return true;
         }
 
         void Complete(const TActorContext&) override {
-            if (TrashDeleted.empty()) {
-                Self->Data->OnCommitConfirmedGC(Channel, GroupId);
+            if (Index == TrashDeleted.size()) {
+                Self->Data->OnCommitConfirmedGC(Channel, GroupId, std::move(TrashDeleted));
             } else { // resume transaction
-                Self->Data->ExecuteConfirmGC(Channel, GroupId, std::move(TrashDeleted), ConfirmedGenStep);
+                Self->Data->ExecuteConfirmGC(Channel, GroupId, std::move(TrashDeleted), Index, ConfirmedGenStep);
             }
         }
     };
 
-    void TData::ExecuteConfirmGC(ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted, TGenStep confirmedGenStep) {
-        Self->Execute(std::make_unique<TTxConfirmGC>(Self, channel, groupId, std::move(trashDeleted), confirmedGenStep));
+    void TData::ExecuteConfirmGC(ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted, size_t index,
+            TGenStep confirmedGenStep) {
+        Self->Execute(std::make_unique<TTxConfirmGC>(Self, channel, groupId, std::move(trashDeleted), index,
+            confirmedGenStep));
     }
 
 } // NKikimr::NBlobDepot
diff --git a/ydb/core/blob_depot/data_trash.cpp b/ydb/core/blob_depot/data_trash.cpp
index 001928a281..79c56c784c 100644
--- a/ydb/core/blob_depot/data_trash.cpp
+++ b/ydb/core/blob_depot/data_trash.cpp
@@ -193,7 +193,7 @@ namespace NKikimr::NBlobDepot {
         if (ev->Get()->Status == NKikimrProto::OK) {
             Y_ABORT_UNLESS(record.CollectGarbageRequestInFlight);
             record.OnSuccessfulCollect(this);
-            ExecuteConfirmGC(record.Channel, record.GroupId, std::exchange(record.TrashInFlight, {}),
+            ExecuteConfirmGC(record.Channel, record.GroupId, std::exchange(record.TrashInFlight, {}), 0,
                 record.LastConfirmedGenStep);
         } else {
             record.TrashInFlight.clear();
@@ -201,8 +201,62 @@ namespace NKikimr::NBlobDepot {
         }
     }
 
-    void TData::OnCommitConfirmedGC(ui8 channel, ui32 groupId) {
+    void TData::OnCommitConfirmedGC(ui8 channel, ui32 groupId, std::vector<TLogoBlobID> trashDeleted) {
         TRecordsPerChannelGroup& record = GetRecordsPerChannelGroup(channel, groupId);
+
+        const TTabletStorageInfo *info = Self->Info();
+        Y_ABORT_UNLESS(info);
+        const TTabletChannelInfo *ch = info->ChannelInfo(channel);
+        Y_ABORT_UNLESS(ch);
+        auto& history = ch->History;
+        Y_ABORT_UNLESS(!history.empty());
+
+        const ui64 tabletId = Self->TabletID();
+
+        ui32 prevGenerationBegin = 0;
+        ui32 prevGenerationEnd = 0;
+
+        Y_DEBUG_ABORT_UNLESS(std::is_sorted(trashDeleted.begin(), trashDeleted.end()));
+
+        for (const TLogoBlobID& id : trashDeleted) {
+            Y_ABORT_UNLESS(id.TabletID() == tabletId);
+            Y_ABORT_UNLESS(id.Channel() == channel);
+
+            const ui32 generation = id.Generation();
+            if (prevGenerationBegin <= generation && generation < prevGenerationEnd) {
+                // the range was already processed, skip
+            } else if (history.back().FromGeneration <= generation) {
+                // in current generation, skip; this is also the last range, as the prefix is equal for all items
+                // and they are in sorted order -- it's safe to quit now
+                break;
+            } else {
+                auto it = std::upper_bound(history.begin(), history.end(), generation,
+                    TTabletChannelInfo::THistoryEntry::TCmp());
+                Y_ABORT_UNLESS(it != history.end());
+                prevGenerationEnd = it->FromGeneration;
+                Y_ABORT_UNLESS(it != history.begin());
+                prevGenerationBegin = std::prev(it)->FromGeneration;
+                Y_ABORT_UNLESS(prevGenerationBegin <= generation && generation < prevGenerationEnd);
+
+                TLogoBlobID min(tabletId, prevGenerationBegin, 0, channel, 0, 0);
+                TLogoBlobID max(tabletId, prevGenerationEnd - 1, Max<ui32>(), channel, TLogoBlobID::MaxBlobSize,
+                    TLogoBlobID::MaxCookie, TLogoBlobID::MaxPartId, TLogoBlobID::MaxCrcMode);
+
+                if (record.Used.lower_bound(min) != record.Used.upper_bound(max)) {
+                    // we have some used records in this range, skip it
+                } else if (record.Trash.lower_bound(min) != record.Trash.upper_bound(max)) {
+                    // we have some still undeleted trash in this range, skip it too
+                } else if (AlreadyCutHistory.emplace(channel, prevGenerationBegin).second) {
+                    auto ev = std::make_unique<TEvTablet::TEvCutTabletHistory>();
+                    auto& record = ev->Record;
+                    record.SetTabletID(tabletId);
+                    record.SetChannel(channel);
+                    record.SetFromGeneration(prevGenerationBegin);
+                    Self->Send(MakeLocalID(Self->SelfId().NodeId()), ev.release());
+                }
+            }
+        }
+
         record.ClearInFlight(this);
     }
 
diff --git a/ydb/core/blob_depot/defs.h b/ydb/core/blob_depot/defs.h
index 6e201abe90..bd9fbc66dd 100644
--- a/ydb/core/blob_depot/defs.h
+++ b/ydb/core/blob_depot/defs.h
@@ -8,6 +8,7 @@
 #include <ydb/core/blobstorage/testing/group_overseer/group_overseer.h>
 #include <ydb/core/blobstorage/vdisk/common/vdisk_events.h>
 #include <ydb/core/engine/minikql/flat_local_tx_factory.h>
+#include <ydb/core/mind/local.h>
 #include <ydb/core/node_whiteboard/node_whiteboard.h>
 #include <ydb/core/tablet/tablet_counters_protobuf.h>
 #include <ydb/core/tablet_flat/tablet_flat_executed.h>