KIKIMR-20009: fix race condition on blobs removing (#1118)

ivanmorozov333 · web-flow · commit 680271ea9e02 · 2024-01-18T15:10:45.000+03:00
diff --git a/ydb/core/tx/columnshard/blob_manager.cpp b/ydb/core/tx/columnshard/blob_manager.cpp
@@ -346,12 +346,15 @@ void TBlobManager::DoSaveBlobBatch(TBlobBatch&& blobBatch, IBlobManagerDb& db) {
     blobBatch.BatchInfo->GenStepRef.Reset();
 }
 
-void TBlobManager::DeleteBlob(const TUnifiedBlobId& blobId, IBlobManagerDb& db) {
-    ++CountersUpdate.BlobsDeleted;
-
+void TBlobManager::DeleteBlobOnExecute(const TUnifiedBlobId& blobId, IBlobManagerDb& db) {
     // Persist deletion intent
     db.AddBlobToDelete(blobId);
-    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("to_delete", blobId);
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("to_delete_on_execute", blobId);
+}
+
+void TBlobManager::DeleteBlobOnComplete(const TUnifiedBlobId& blobId) {
+    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("to_delete_on_complete", blobId);
+    ++CountersUpdate.BlobsDeleted;
 
     // Check if the deletion needs to be delayed until the blob is no longer
     // used by in-flight requests
diff --git a/ydb/core/tx/columnshard/blob_manager.h b/ydb/core/tx/columnshard/blob_manager.h
@@ -90,7 +90,8 @@ class IBlobManager {
     }
 
     // Deletes the blob that was previously permanently saved
-    virtual void DeleteBlob(const TUnifiedBlobId& blobId, IBlobManagerDb& db) = 0;
+    virtual void DeleteBlobOnExecute(const TUnifiedBlobId& blobId, IBlobManagerDb& db) = 0;
+    virtual void DeleteBlobOnComplete(const TUnifiedBlobId& blobId) = 0;
 };
 
 // An interface for exporting and caching exported blobs out of ColumnShard index to external storages like S3.
@@ -218,7 +219,8 @@ class TBlobManager : public IBlobManager, public NOlap::TCommonBlobsTracker {
 
     // Implementation of IBlobManager interface
     TBlobBatch StartBlobBatch(ui32 channel = BLOB_CHANNEL) override;
-    void DeleteBlob(const TUnifiedBlobId& blobId, IBlobManagerDb& db) override;
+    void DeleteBlobOnExecute(const TUnifiedBlobId& blobId, IBlobManagerDb& db) override;
+    void DeleteBlobOnComplete(const TUnifiedBlobId& blobId) override;
 private:
     TGenStep FindNewGCBarrier();
 
diff --git a/ydb/core/tx/columnshard/blobs_action/abstract/action.h b/ydb/core/tx/columnshard/blobs_action/abstract/action.h
@@ -64,12 +64,12 @@ class TStorageAction {
         }
     }
 
-    void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self) {
+    void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self, const bool success) {
         if (Removing) {
-            Removing->OnCompleteTxAfterRemoving(self);
+            Removing->OnCompleteTxAfterRemoving(self, success);
         }
         if (Writing) {
-            Writing->OnCompleteTxAfterWrite(self);
+            Writing->OnCompleteTxAfterWrite(self, success);
         }
     }
 };
@@ -150,9 +150,9 @@ class TBlobsAction {
         }
     }
 
-    void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self) {
+    void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self, const bool success) {
         for (auto&& i : StorageActions) {
-            i.second.OnCompleteTxAfterAction(self);
+            i.second.OnCompleteTxAfterAction(self, success);
         }
     }
 
diff --git a/ydb/core/tx/columnshard/blobs_action/abstract/remove.h b/ydb/core/tx/columnshard/blobs_action/abstract/remove.h
@@ -20,7 +20,7 @@ class IBlobsDeclareRemovingAction: public ICommonBlobsAction {
 protected:
     virtual void DoDeclareRemove(const TUnifiedBlobId& blobId) = 0;
     virtual void DoOnExecuteTxAfterRemoving(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) = 0;
-    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& self) = 0;
+    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& self, const bool success) = 0;
 public:
     IBlobsDeclareRemovingAction(const TString& storageId)
         : TBase(storageId)
@@ -36,8 +36,8 @@ class IBlobsDeclareRemovingAction: public ICommonBlobsAction {
     void OnExecuteTxAfterRemoving(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) {
         return DoOnExecuteTxAfterRemoving(self, dbBlobs, success);
     }
-    void OnCompleteTxAfterRemoving(NColumnShard::TColumnShard& self) {
-        return DoOnCompleteTxAfterRemoving(self);
+    void OnCompleteTxAfterRemoving(NColumnShard::TColumnShard& self, const bool success) {
+        return DoOnCompleteTxAfterRemoving(self, success);
     }
 };
 
diff --git a/ydb/core/tx/columnshard/blobs_action/abstract/write.h b/ydb/core/tx/columnshard/blobs_action/abstract/write.h
@@ -31,7 +31,7 @@ class IBlobsWritingAction: public ICommonBlobsAction {
     virtual void DoOnBlobWriteResult(const TUnifiedBlobId& blobId, const NKikimrProto::EReplyStatus status) = 0;
 
     virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) = 0;
-    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& self) = 0;
+    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& self, const bool success) = 0;
 
     virtual TUnifiedBlobId AllocateNextBlobId(const TString& data) = 0;
 public:
@@ -79,8 +79,8 @@ class IBlobsWritingAction: public ICommonBlobsAction {
         return DoOnExecuteTxAfterWrite(self, dbBlobs, success);
     }
 
-    void OnCompleteTxAfterWrite(NColumnShard::TColumnShard& self) {
-        return DoOnCompleteTxAfterWrite(self);
+    void OnCompleteTxAfterWrite(NColumnShard::TColumnShard& self, const bool success) {
+        return DoOnCompleteTxAfterWrite(self, success);
     }
 
     void SendWriteBlobRequest(const TString& data, const TUnifiedBlobId& blobId);
diff --git a/ydb/core/tx/columnshard/blobs_action/bs/remove.h b/ydb/core/tx/columnshard/blobs_action/bs/remove.h
@@ -18,12 +18,16 @@ class TDeclareRemovingAction: public IBlobsDeclareRemovingAction {
     virtual void DoOnExecuteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) {
         if (success) {
             for (auto&& i : GetDeclaredBlobs()) {
-                Manager->DeleteBlob(i, dbBlobs);
+                Manager->DeleteBlobOnExecute(i, dbBlobs);
             }
         }
     }
-    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/) {
-
+    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/, const bool success) {
+        if (success) {
+            for (auto&& i : GetDeclaredBlobs()) {
+                Manager->DeleteBlobOnComplete(i);
+            }
+        }
     }
 public:
     TDeclareRemovingAction(const TString& storageId, NColumnShard::TBlobManager& manager)
diff --git a/ydb/core/tx/columnshard/blobs_action/bs/write.h b/ydb/core/tx/columnshard/blobs_action/bs/write.h
@@ -29,7 +29,7 @@ class TWriteAction: public IBlobsWritingAction {
     }
 
     virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) override;
-    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /*self*/) override {
+    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /*self*/, const bool /*success*/) override {
 
     }
 public:
diff --git a/ydb/core/tx/columnshard/blobs_action/memory.h b/ydb/core/tx/columnshard/blobs_action/memory.h
@@ -71,7 +71,7 @@ class TMemoryWriteAction: public IBlobsWritingAction {
     virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& /*self*/, NColumnShard::TBlobManagerDb& /*dbBlobs*/, const bool /*success*/) override {
 
     }
-    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /*self*/) override {
+    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /*self*/, const bool /*success*/) override {
 
     }
 public:
@@ -106,7 +106,7 @@ class TMemoryDeclareRemovingAction: public IBlobsDeclareRemovingAction {
             Storage->DeclareDataForRemove(i);
         }
     }
-    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/) {
+    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/, const bool /*success*/) {
 
     }
 public:
diff --git a/ydb/core/tx/columnshard/blobs_action/tier/remove.h b/ydb/core/tx/columnshard/blobs_action/tier/remove.h
@@ -20,6 +20,12 @@ class TDeclareRemovingAction: public IBlobsDeclareRemovingAction {
         if (success) {
             for (auto&& i : GetDeclaredBlobs()) {
                 dbBlobs.AddTierBlobToDelete(GetStorageId(), i);
+            }
+        }
+    }
+    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/, const bool success) {
+        if (success) {
+            for (auto&& i : GetDeclaredBlobs()) {
                 if (GCInfo->IsBlobInUsage(i)) {
                     Y_ABORT_UNLESS(GCInfo->MutableBlobsToDeleteInFuture().emplace(i).second);
                 } else {
@@ -28,9 +34,6 @@ class TDeclareRemovingAction: public IBlobsDeclareRemovingAction {
             }
         }
     }
-    virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /*self*/) {
-
-    }
 public:
     TDeclareRemovingAction(const TString& storageId, const std::shared_ptr<TGCInfo>& gcInfo)
         : TBase(storageId)
diff --git a/ydb/core/tx/columnshard/blobs_action/tier/write.h b/ydb/core/tx/columnshard/blobs_action/tier/write.h
@@ -27,7 +27,7 @@ class TWriteAction: public IBlobsWritingAction {
     }
 
     virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) override;
-    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /*self*/) override {
+    virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /*self*/, const bool /*success*/) override {
 
     }
 public:
diff --git a/ydb/core/tx/columnshard/blobs_action/transaction/tx_gc_insert_table.cpp b/ydb/core/tx/columnshard/blobs_action/transaction/tx_gc_insert_table.cpp
@@ -21,7 +21,7 @@ bool TTxInsertTableCleanup::Execute(TTransactionContext& txc, const TActorContex
 }
 void TTxInsertTableCleanup::Complete(const TActorContext& /*ctx*/) {
     Y_ABORT_UNLESS(BlobsAction);
-    BlobsAction->OnCompleteTxAfterRemoving(*Self);
+    BlobsAction->OnCompleteTxAfterRemoving(*Self, true);
     Self->EnqueueBackgroundActivities();
 }
 
diff --git a/ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp b/ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp
@@ -102,10 +102,10 @@ void TTxWrite::Complete(const TActorContext& ctx) {
     const auto now = TMonotonic::Now();
     const NOlap::TWritingBuffer& buffer = PutBlobResult->Get()->MutableWritesBuffer();
     for (auto&& i : buffer.GetAddActions()) {
-        i->OnCompleteTxAfterWrite(*Self);
+        i->OnCompleteTxAfterWrite(*Self, true);
     }
     for (auto&& i : buffer.GetRemoveActions()) {
-        i->OnCompleteTxAfterRemoving(*Self);
+        i->OnCompleteTxAfterRemoving(*Self, true);
     }
     AFL_VERIFY(buffer.GetAggregations().size() == Results.size());
     for (ui32 i = 0; i < buffer.GetAggregations().size(); ++i) {
diff --git a/ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp b/ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp
@@ -64,7 +64,7 @@ void TTxWriteIndex::Complete(const TActorContext& ctx) {
         Self->EnqueueBackgroundActivities(false, TriggerActivity);
     }
 
-    changes->MutableBlobsAction().OnCompleteTxAfterAction(*Self);
+    changes->MutableBlobsAction().OnCompleteTxAfterAction(*Self, Ev->Get()->GetPutStatus() == NKikimrProto::OK);
     NYDBTest::TControllers::GetColumnShardController()->OnWriteIndexComplete(Self->TabletID(), changes->TypeString());
 }
 

Original file line number	Diff line number	Diff line change
`@@ -64,12 +64,12 @@ class TStorageAction {`
`64`	`64`	`}`
`65`	`65`	`}`
`66`	`66`
`67`		`- void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self) {`
	`67`	`+ void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self, const bool success) {`
`68`	`68`	`if (Removing) {`
`69`		`- Removing->OnCompleteTxAfterRemoving(self);`
	`69`	`+ Removing->OnCompleteTxAfterRemoving(self, success);`
`70`	`70`	`}`
`71`	`71`	`if (Writing) {`
`72`		`- Writing->OnCompleteTxAfterWrite(self);`
	`72`	`+ Writing->OnCompleteTxAfterWrite(self, success);`
`73`	`73`	`}`
`74`	`74`	`}`
`75`	`75`	`};`
`@@ -150,9 +150,9 @@ class TBlobsAction {`
`150`	`150`	`}`
`151`	`151`	`}`
`152`	`152`
`153`		`- void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self) {`
	`153`	`+ void OnCompleteTxAfterAction(NColumnShard::TColumnShard& self, const bool success) {`
`154`	`154`	`for (auto&& i : StorageActions) {`
`155`		`- i.second.OnCompleteTxAfterAction(self);`
	`155`	`+ i.second.OnCompleteTxAfterAction(self, success);`
`156`	`156`	`}`
`157`	`157`	`}`
`158`	`158`
Original file line number	Diff line number	Diff line change
`@@ -18,12 +18,16 @@ class TDeclareRemovingAction: public IBlobsDeclareRemovingAction {`
`18`	`18`	`virtual void DoOnExecuteTxAfterRemoving(NColumnShard::TColumnShard& /self/, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) {`
`19`	`19`	`if (success) {`
`20`	`20`	`for (auto&& i : GetDeclaredBlobs()) {`
`21`		`- Manager->DeleteBlob(i, dbBlobs);`
	`21`	`+ Manager->DeleteBlobOnExecute(i, dbBlobs);`
`22`	`22`	`}`
`23`	`23`	`}`
`24`	`24`	`}`
`25`		`- virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /self/) {`
`26`		`-`
	`25`	`+ virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /self/, const bool success) {`
	`26`	`+ if (success) {`
	`27`	`+ for (auto&& i : GetDeclaredBlobs()) {`
	`28`	`+ Manager->DeleteBlobOnComplete(i);`
	`29`	`+ }`
	`30`	`+ }`
`27`	`31`	`}`
`28`	`32`	`public:`
`29`	`33`	`TDeclareRemovingAction(const TString& storageId, NColumnShard::TBlobManager& manager)`
Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,7 @@ class TWriteAction: public IBlobsWritingAction {`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	`virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) override;`
`32`		`- virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /self/) override {`
	`32`	`+ virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /self/, const bool /success/) override {`
`33`	`33`
`34`	`34`	`}`
`35`	`35`	`public:`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ class TMemoryWriteAction: public IBlobsWritingAction {`
`71`	`71`	`virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& /self/, NColumnShard::TBlobManagerDb& /dbBlobs/, const bool /success/) override {`
`72`	`72`
`73`	`73`	`}`
`74`		`- virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /self/) override {`
	`74`	`+ virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /self/, const bool /success/) override {`
`75`	`75`
`76`	`76`	`}`
`77`	`77`	`public:`
`@@ -106,7 +106,7 @@ class TMemoryDeclareRemovingAction: public IBlobsDeclareRemovingAction {`
`106`	`106`	`Storage->DeclareDataForRemove(i);`
`107`	`107`	`}`
`108`	`108`	`}`
`109`		`- virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /self/) {`
	`109`	`+ virtual void DoOnCompleteTxAfterRemoving(NColumnShard::TColumnShard& /self/, const bool /success/) {`
`110`	`110`
`111`	`111`	`}`
`112`	`112`	`public:`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ class TWriteAction: public IBlobsWritingAction {`
`27`	`27`	`}`
`28`	`28`
`29`	`29`	`virtual void DoOnExecuteTxAfterWrite(NColumnShard::TColumnShard& self, NColumnShard::TBlobManagerDb& dbBlobs, const bool success) override;`
`30`		`- virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /self/) override {`
	`30`	`+ virtual void DoOnCompleteTxAfterWrite(NColumnShard::TColumnShard& /self/, const bool /success/) override {`
`31`	`31`
`32`	`32`	`}`
`33`	`33`	`public:`
Original file line number	Diff line number	Diff line change
`@@ -21,7 +21,7 @@ bool TTxInsertTableCleanup::Execute(TTransactionContext& txc, const TActorContex`
`21`	`21`	`}`
`22`	`22`	`void TTxInsertTableCleanup::Complete(const TActorContext& /ctx/) {`
`23`	`23`	`Y_ABORT_UNLESS(BlobsAction);`
`24`		`- BlobsAction->OnCompleteTxAfterRemoving(*Self);`
	`24`	`+ BlobsAction->OnCompleteTxAfterRemoving(*Self, true);`
`25`	`25`	`Self->EnqueueBackgroundActivities();`
`26`	`26`	`}`
`27`	`27`
Original file line number	Diff line number	Diff line change
`@@ -102,10 +102,10 @@ void TTxWrite::Complete(const TActorContext& ctx) {`
`102`	`102`	`const auto now = TMonotonic::Now();`
`103`	`103`	`const NOlap::TWritingBuffer& buffer = PutBlobResult->Get()->MutableWritesBuffer();`
`104`	`104`	`for (auto&& i : buffer.GetAddActions()) {`
`105`		`- i->OnCompleteTxAfterWrite(*Self);`
	`105`	`+ i->OnCompleteTxAfterWrite(*Self, true);`
`106`	`106`	`}`
`107`	`107`	`for (auto&& i : buffer.GetRemoveActions()) {`
`108`		`- i->OnCompleteTxAfterRemoving(*Self);`
	`108`	`+ i->OnCompleteTxAfterRemoving(*Self, true);`
`109`	`109`	`}`
`110`	`110`	`AFL_VERIFY(buffer.GetAggregations().size() == Results.size());`
`111`	`111`	`for (ui32 i = 0; i < buffer.GetAggregations().size(); ++i) {`
Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ void TTxWriteIndex::Complete(const TActorContext& ctx) {`
`64`	`64`	`Self->EnqueueBackgroundActivities(false, TriggerActivity);`
`65`	`65`	`}`
`66`	`66`
`67`		`- changes->MutableBlobsAction().OnCompleteTxAfterAction(*Self);`
	`67`	`+ changes->MutableBlobsAction().OnCompleteTxAfterAction(*Self, Ev->Get()->GetPutStatus() == NKikimrProto::OK);`
`68`	`68`	`NYDBTest::TControllers::GetColumnShardController()->OnWriteIndexComplete(Self->TabletID(), changes->TypeString());`
`69`	`69`	`}`
`70`	`70`