Data cleanup fixes (ydb-platform#16627)

Fixes for data cleanup edge cases:
- DataShard: clean readsets in DataCleanup (ydb-platform#15438)
- DataShard and SchemeShard: handle borrowed parts in data erasure (ydb-platform#15451)

Author: lex007in

ydb/core/protos/tx_datashard.proto

@@ -2280,8 +2280,10 @@ message TEvForceDataCleanup {
 // Intermediate requests and corresponding TEvForceDataCleanupResult's may be skipped.
 message TEvForceDataCleanupResult {
     enum EStatus {
-        OK = 0;
-        FAILED = 1;
+        UNKNOWN = 0;
+        OK = 1;
+        WRONG_SHARD_STATE = 2;
+        BORROWED = 3;
     };
     optional uint64 DataCleanupGeneration = 1; // from corresponding request (or greater)
     optional uint64 TabletId = 2;

ydb/core/testlib/storage_helpers.cpp

@@ -0,0 +1,28 @@
+#include "storage_helpers.h"
+
+#include <ydb/core/blobstorage/dsproxy/mock/model.h>
+
+namespace NKikimr {
+    int CountBlobsWithSubstring(ui64 tabletId, const TVector<TIntrusivePtr<NFake::TProxyDS>>& proxyDSs, const TString& substring) {
+        int res = 0;
+        for (const auto& proxyDS : proxyDSs) {
+            for (const auto& [id, blob] : proxyDS->AllMyBlobs()) {
+                if (id.TabletID() == tabletId && !blob.DoNotKeep && blob.Buffer.ConvertToString().Contains(substring)) {
+                    ++res;
+                }
+            }
+        }
+        return res;
+    }
+
+    bool BlobStorageContains(const TVector<TIntrusivePtr<NFake::TProxyDS>>& proxyDSs, const TString& value) {
+        for (const auto& proxyDS : proxyDSs) {
+            for (const auto& [id, blob] : proxyDS->AllMyBlobs()) {
+                if (!blob.DoNotKeep && blob.Buffer.ConvertToString().Contains(value)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+} // namespace NKikimr

ydb/core/testlib/storage_helpers.h

@@ -0,0 +1,8 @@
+#pragma once
+
+#include <ydb/core/blobstorage/dsproxy/mock/dsproxy_mock.h>
+
+namespace NKikimr {
+    int CountBlobsWithSubstring(ui64 tabletId, const TVector<TIntrusivePtr<NFake::TProxyDS>>& proxyDSs, const TString& substring);
+    bool BlobStorageContains(const TVector<TIntrusivePtr<NFake::TProxyDS>>& proxyDSs, const TString& value);
+} // namespace NKikimr

ydb/core/testlib/test_client.h

@@ -109,7 +109,6 @@ namespace Tests {
         using TControls = NKikimrConfig::TImmediateControlsConfig;
         using TLoggerInitializer = std::function<void (TTestActorRuntime&)>;
         using TStoragePoolKinds = TDomainsInfo::TDomain::TStoragePoolKinds;
-        using TProxyDSPtr = TIntrusivePtr<NFake::TProxyDS>;
 
         ui16 Port;
         ui16 GrpcPort = 0;
@@ -170,7 +169,7 @@ namespace Tests {
         TString ServerCertFilePath;
         bool Verbose = true;
         bool UseSectorMap = false;
-        TVector<TProxyDSPtr> ProxyDSMocks;
+        TVector<TIntrusivePtr<NFake::TProxyDS>> ProxyDSMocks;
 
         std::function<IActor*(const TTicketParserSettings&)> CreateTicketParser = NKikimr::CreateTicketParser;
         std::shared_ptr<TGrpcServiceFactory> GrpcServiceFactory;
@@ -261,7 +260,7 @@ namespace Tests {
             return *this;
         }
 
-        TServerSettings& SetProxyDSMocks(const TVector<TProxyDSPtr>& proxyDSMocks) {
+        TServerSettings& SetProxyDSMocks(const TVector<TIntrusivePtr<NFake::TProxyDS>>& proxyDSMocks) {
             ProxyDSMocks = proxyDSMocks;
             return *this;
         }

ydb/core/testlib/ya.make

@@ -10,6 +10,7 @@ SRCS(
     fake_scheme_shard.h
     minikql_compile.h
     mock_pq_metacache.h
+    storage_helpers.cpp
     tablet_flat_dummy.cpp
     tablet_helpers.cpp
     tablet_helpers.h

ydb/core/tx/datashard/datashard__data_cleanup.cpp

@@ -26,7 +26,19 @@ class TDataShard::TTxDataCleanup : public NTabletFlatExecutor::TTransactionBase<
             Response = std::make_unique<TEvDataShard::TEvForceDataCleanupResult>(
                 record.GetDataCleanupGeneration(),
                 Self->TabletID(),
-                NKikimrTxDataShard::TEvForceDataCleanupResult::FAILED);
+                NKikimrTxDataShard::TEvForceDataCleanupResult::WRONG_SHARD_STATE);
+            return true;
+        }
+
+        if (Self->Executor()->HasLoanedParts()) {
+            LOG_WARN_S(ctx, NKikimrServices::TX_DATASHARD,
+                "DataCleanup of tablet# " << Self->TabletID()
+                << ": has borrowed parts"
+                << ", requested from " << Ev->Sender);
+            Response = std::make_unique<TEvDataShard::TEvForceDataCleanupResult>(
+                record.GetDataCleanupGeneration(),
+                Self->TabletID(),
+                NKikimrTxDataShard::TEvForceDataCleanupResult::BORROWED);
             return true;
         }
 
@@ -55,6 +67,8 @@ class TDataShard::TTxDataCleanup : public NTabletFlatExecutor::TTransactionBase<
                 "DataCleanup of tablet# " << Self->TabletID()
                 << ": expired snapshots removed");
         }
+        Self->OutReadSets.Cleanup(db, ctx);
+
         Self->Executor()->CleanupData(Ev->Get()->Record.GetDataCleanupGeneration());
         Self->DataCleanupWaiters.insert({Ev->Get()->Record.GetDataCleanupGeneration(), Ev->Sender});
         return true;