Fix Trash table loading (#30273)

Author: alexvru

ydb/core/blob_depot/data.cpp

@@ -201,6 +201,8 @@ namespace NKikimr::NBlobDepot {
                 Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_TOTAL_STORED_DATA_SIZE] = TotalStoredDataSize;
 
                 InFlightTrashBlobs.emplace(cookie, id);
+                const bool inserted = AllInFlightTrashBlobs.insert(id).second;
+                Y_ABORT_UNLESS(inserted);
                 db.Table<Schema::Trash>().Key(id.AsBinaryString()).Update();
                 return false; // keep this blob in deletion queue
             };
@@ -462,8 +464,13 @@ namespace NKikimr::NBlobDepot {
     }
 
     void TData::AddTrashOnLoad(TLogoBlobID id) {
+        if (AllInFlightTrashBlobs.contains(id)) {
+            return; // we're trying to add just inserted item, ignore it
+        }
         auto& record = GetRecordsPerChannelGroup(id);
-        record.Trash.insert(id);
+        if (const auto [it, inserted] = record.Trash.insert(id); !inserted) {
+            return; // the same situation: this item has just been inserted into the set
+        }
         AccountBlob(id, true);
         TotalStoredTrashSize += id.BlobSize();
         Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_TOTAL_STORED_TRASH_SIZE] = TotalStoredTrashSize;
@@ -646,7 +653,8 @@ namespace NKikimr::NBlobDepot {
     void TData::TRecordsPerChannelGroup::MoveToTrash(TData *self, TLogoBlobID id) {
         const auto usedIt = Used.find(id);
         Y_ABORT_UNLESS(usedIt != Used.end());
-        Trash.insert(Used.extract(usedIt));
+        const bool inserted = Trash.insert(Used.extract(usedIt)).inserted;
+        Y_DEBUG_ABORT_UNLESS(inserted);
         self->TotalStoredTrashSize += id.BlobSize();
         self->Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_TOTAL_STORED_TRASH_SIZE] = self->TotalStoredTrashSize;
     }
@@ -784,6 +792,7 @@ namespace NKikimr::NBlobDepot {
         for (const auto& [cookie, id] : InFlightTrashBlobs) {
             const bool inserted = refcountBlobs.try_emplace(id).second;
             Y_ABORT_UNLESS(inserted);
+            Y_ABORT_UNLESS(AllInFlightTrashBlobs.contains(id));
         }
 
         for (const auto& [cookie, locator] : InFlightTrashS3) {

ydb/core/blob_depot/data.h

@@ -479,6 +479,7 @@ namespace NKikimr::NBlobDepot {
 
         THashMultiMap<void*, TLogoBlobID> InFlightTrashBlobs; // being committed, but not yet confirmed
         THashMultiMap<void*, TS3Locator> InFlightTrashS3; // being committed, but not yet confirmed
+        THashSet<TLogoBlobID> AllInFlightTrashBlobs;
 
         class TTxIssueGC;
         class TTxConfirmGC;

ydb/core/blob_depot/data_trash.cpp

@@ -13,6 +13,7 @@ namespace NKikimr::NBlobDepot {
             record.MoveToTrash(this, it->second);
             records.insert(&record);
             InFlightTrashSize -= it->second.BlobSize();
+            AllInFlightTrashBlobs.erase(it->second);
         }
         InFlightTrashBlobs.erase(first, last);