Reuse S3 locators to prevent false trash accumulation (#15390)

Author: alexvru

ydb/core/blob_depot/agent/agent_impl.h

@@ -194,6 +194,7 @@ namespace NKikimr::NBlobDepot {
         TActorId PipeId;
         TActorId PipeServerId;
         bool IsConnected = false;
+        ui64 ConnectionInstance = 0;
 
         NMonitoring::TDynamicCounterPtr AgentCounters;
 

ydb/core/blob_depot/agent/comm.cpp

@@ -155,6 +155,8 @@ namespace NKikimr::NBlobDepot {
     }
 
     void TBlobDepotAgent::OnDisconnect() {
+        ++ConnectionInstance;
+
         while (!TabletRequestInFlight.empty()) {
             auto node = TabletRequestInFlight.extract(TabletRequestInFlight.begin());
             auto& requestInFlight = node.value();

ydb/core/blob_depot/agent/storage_put.cpp

@@ -18,6 +18,7 @@ namespace NKikimr::NBlobDepot {
             TBlobSeqId BlobSeqId;
             std::optional<TS3Locator> LocatorInFlight;
             TActorId WriterActorId;
+            ui64 ConnectionInstanceOnStart;
 
             struct TLifetimeToken {};
             std::shared_ptr<TLifetimeToken> LifetimeToken;
@@ -410,6 +411,8 @@ namespace NKikimr::NBlobDepot {
                             .AddMetadata("key", Request.Id.ToString()),
                         Request.Buffer.ExtractUnderlyingContainerOrCopy<TString>()),
                     IEventHandle::FlagTrackDelivery));
+
+                ConnectionInstanceOnStart = Agent.ConnectionInstance;
             }
 
             void OnPutS3ObjectResponse(std::optional<TString>&& error) {
@@ -418,6 +421,11 @@ namespace NKikimr::NBlobDepot {
 
                 WriterActorId = {};
 
+                if (ConnectionInstanceOnStart != Agent.ConnectionInstance) {
+                    error = "BlobDepot tablet disconnected";
+                    LocatorInFlight.reset(); // prevent discarding this locator
+                }
+
                 if (error) {
                     ++*Agent.S3PutsError;
 

ydb/core/blob_depot/op_commit_blob_seq.cpp

@@ -75,13 +75,8 @@ namespace NKikimr::NBlobDepot {
                         if (item.HasS3Locator()) {
                             const auto& locator = TS3Locator::FromProto(item.GetS3Locator());
                             const size_t numErased = agent.S3WritesInFlight.erase(locator);
-                            if (locator.Generation < generation) {
-                                Y_ABORT_UNLESS(!numErased);
-                                Self->Data->AddToS3Trash(locator, txc, this);
-                            } else {
-                                Y_ABORT_UNLESS(numErased == 1);
-                                Self->S3Manager->AddTrashToCollect(locator);
-                            }
+                            Y_ABORT_UNLESS(numErased);
+                            Self->S3Manager->AddTrashToCollect(locator);
                         }
                     };
 
@@ -244,11 +239,10 @@ namespace NKikimr::NBlobDepot {
         }
 
         for (const auto& item : record.GetS3Locators()) {
-            if (const auto& locator = TS3Locator::FromProto(item); locator.Generation == generation) {
-                const size_t numErased = agent.S3WritesInFlight.erase(locator);
-                Y_ABORT_UNLESS(numErased == 1);
-                S3Manager->AddTrashToCollect(locator);
-            }
+            const auto& locator = TS3Locator::FromProto(item);
+            const size_t numErased = agent.S3WritesInFlight.erase(locator);
+            Y_ABORT_UNLESS(numErased == 1);
+            S3Manager->AddTrashToCollect(locator);
         }
     }
 

ydb/core/blob_depot/s3_write.cpp

@@ -53,7 +53,7 @@ namespace NKikimr::NBlobDepot {
                     locator.ToProto(responseItem->MutableS3Locator());
 
                     // we put it here until operation is completed; if tablet restarts and operation fails, then this
-                    // key will be deleted
+                    // key will be deleted; we also rewrite spoiled locator because Len is different
                     db.Table<Schema::TrashS3>().Key(locator.Generation, locator.KeyId).Update<Schema::TrashS3::Len>(locator.Len);
 
                     const bool inserted = agent.S3WritesInFlight.insert(locator).second;
@@ -98,6 +98,14 @@ namespace NKikimr::NBlobDepot {
     };
 
     TS3Locator TS3Manager::AllocateS3Locator(ui32 len) {
+        if (!DeleteQueue.empty()) {
+            TS3Locator res = DeleteQueue.front();
+            DeleteQueue.pop_front();
+            Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_TOTAL_S3_TRASH_OBJECTS] = --TotalS3TrashObjects;
+            Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_TOTAL_S3_TRASH_SIZE] = TotalS3TrashSize -= res.Len;
+            res.Len = len;
+            return res;
+        }
         return {
             .Len = len,
             .Generation = Self->Executor()->Generation(),

ydb/core/blob_depot/types.h

@@ -153,7 +153,7 @@ namespace NKikimr::NBlobDepot {
         }
 
         TString MakeObjectName(const TString& basePath) const {
-            const size_t hash = THash()(*this);
+            const size_t hash = MultiHash(Generation, KeyId);
             const size_t a = hash % 36;
             const size_t b = hash / 36 % 36;
             static const char vec[] = "0123456789abcdefghijklmnopqrstuvwxyz";