support CheckIntegrity request in blob depot (#20588)

alexd65536 · web-flow · commit cb76c1dc6f2c · 2025-07-04T13:39:07.000+03:00
diff --git a/ydb/core/base/blobstorage.h b/ydb/core/base/blobstorage.h
@@ -1482,21 +1482,20 @@ struct TEvBlobStorage {
 
         enum EPlacementStatus {
             PS_OK = 1,                      // blob parts are placed according to fail model
-            PS_BLOB_IS_LOST = 2,            // blob is lost/unrecoverable
+            PS_REPLICATION_IN_PROGRESS = 2, // there are missing parts but status may become OK after replication
             PS_UNKNOWN = 3,                 // status is unknown because of missing disks or network problems
-            PS_REPLICATION_IN_PROGRESS = 4, // there are missing parts but status may become OK after replication
-            PS_BLOB_IS_RECOVERABLE = 5,     // blob parts are definitely placed incorrectly or there are missing parts
-                                            // but blob may be recovered
+            PS_BLOB_IS_RECOVERABLE = 4,     // blob parts are definitely placed incorrectly or there are missing parts but blob may be recovered
+            PS_BLOB_IS_LOST = 5,            // blob is lost/unrecoverable
         };
         EPlacementStatus PlacementStatus;
 
         enum EDataStatus {
             DS_OK = 1,      // all data parts contain valid data
-            DS_ERROR = 2,   // some parts definitely contain invalid data
-            DS_UNKNOWN = 3, // status is unknown because of missing disks or network problems
+            DS_UNKNOWN = 2, // status is unknown because of missing disks or network problems
+            DS_ERROR = 3,   // some parts definitely contain invalid data
         };
         EDataStatus DataStatus;
-        TString DataErrorInfo; // textual info about errors in blob data
+        TString DataInfo; // textual info about checks in blob data
 
         std::shared_ptr<TExecutionRelay> ExecutionRelay;
 
@@ -1512,7 +1511,7 @@ struct TEvBlobStorage {
                 << " ErrorReason# " << ErrorReason
                 << " PlacementStatus# " << (int)PlacementStatus
                 << " DataStatus# " << (int)DataStatus
-                << " DataErrorInfo# " << DataErrorInfo
+                << " DataInfo# " << DataInfo
                 << " }";
             return str.Str();
         }
diff --git a/ydb/core/blob_depot/agent/agent_impl.h b/ydb/core/blob_depot/agent/agent_impl.h
@@ -17,6 +17,7 @@ namespace NKikimr::NBlobDepot {
         XX(EvCollectGarbage) \
         XX(EvStatus) \
         XX(EvPatch) \
+        XX(EvCheckIntegrity) \
         // END
 
     class TBlobDepotAgent;
@@ -58,6 +59,7 @@ namespace NKikimr::NBlobDepot {
         bool Error() const { return std::holds_alternative<TError>(Outcome); }
         bool Success() const { return std::holds_alternative<TSuccess>(Outcome); }
         const TResolvedValue *GetResolvedValue() const { return std::get<TSuccess>(Outcome).Value; }
+        TString GetErrorReason() const { return std::get<TError>(Outcome).ErrorReason; }
 
         void Output(IOutputStream& s) const {
             if (auto *success = std::get_if<TSuccess>(&Outcome)) {
@@ -133,7 +135,8 @@ namespace NKikimr::NBlobDepot {
 
             // underlying DS proxy responses
             TEvBlobStorage::TEvGetResult*,
-            TEvBlobStorage::TEvPutResult*
+            TEvBlobStorage::TEvPutResult*,
+            TEvBlobStorage::TEvCheckIntegrityResult*
         >;
 
         static TString ToString(const TResponse& response);
@@ -181,6 +184,14 @@ namespace NKikimr::NBlobDepot {
         }
     };
 
+    struct TCheckOutcome {
+        std::unique_ptr<TEvBlobStorage::TEvCheckIntegrityResult> Result;
+
+        TString ToString() const {
+            return TStringBuilder() << "{Result# " << (Result ? Result->ToString() : "") << "}";
+        }
+    };
+
     class TBlobDepotAgent
         : public TActorBootstrapped<TBlobDepotAgent>
         , public TRequestSender
@@ -270,6 +281,7 @@ namespace NKikimr::NBlobDepot {
 
                 hFunc(TEvBlobStorage::TEvGetResult, HandleOtherResponse);
                 hFunc(TEvBlobStorage::TEvPutResult, HandleOtherResponse);
+                hFunc(TEvBlobStorage::TEvCheckIntegrityResult, HandleOtherResponse);
 
                 ENUMERATE_INCOMING_EVENTS(FORWARD_STORAGE_PROXY)
                 fFunc(TEvBlobStorage::EvAssimilate, HandleAssimilate);
@@ -429,6 +441,7 @@ namespace NKikimr::NBlobDepot {
             virtual void OnIdAllocated(bool /*success*/) {}
             virtual void OnDestroy(bool /*success*/) {}
             virtual void OnPutS3ObjectResponse(std::optional<TString>&& /*error*/) { Y_ABORT(); }
+            virtual void OnCheckIntegrity(TCheckOutcome&& /*outcome*/) {}
 
             NKikimrProto::EReplyStatus CheckBlockForTablet(ui64 tabletId, std::optional<ui32> generation,
                 ui32 *blockedGeneration = nullptr);
@@ -450,11 +463,15 @@ namespace NKikimr::NBlobDepot {
                 std::optional<TEvBlobStorage::TEvGet::TReaderTabletData> ReaderTabletData;
                 TString Key; // the key we are reading -- this is used for retries when we are getting NODATA
             };
+            struct TCheckContext;
 
             bool IssueRead(TReadArg&& arg, TString& error);
             void HandleGetResult(const TRequestContext::TPtr& context, TEvBlobStorage::TEvGetResult& msg);
             void HandleResolveResult(const TRequestContext::TPtr& context, TEvBlobDepot::TEvResolveResult& msg);
 
+            void IssueCheckIntegrity(TReadArg&& arg);
+            void HandleCheckIntegrityResult(const TRequestContext::TPtr& context, TEvBlobStorage::TEvCheckIntegrityResult& msg);
+
         public:
             struct TDeleter {
                 static void Destroy(TQuery *query) { delete query; }
diff --git a/ydb/core/blob_depot/agent/proxy.cpp b/ydb/core/blob_depot/agent/proxy.cpp
@@ -14,6 +14,10 @@ namespace NKikimr::NBlobDepot {
             case TEvBlobStorage::EvGet:
                 static_cast<TEvBlobStorage::TEvGet&>(*event).ExecutionRelay = executionRelay;
                 break;
+
+            case TEvBlobStorage::EvCheckIntegrity:
+                static_cast<TEvBlobStorage::TEvCheckIntegrity&>(*event).ExecutionRelay = executionRelay;
+                break;
         }
 
         const ui64 id = NextOtherRequestId++;
diff --git a/ydb/core/blob_depot/agent/read.cpp b/ydb/core/blob_depot/agent/read.cpp
@@ -58,6 +58,24 @@ namespace NKikimr::NBlobDepot {
         };
     };
 
+    struct TBlobDepotAgent::TQuery::TCheckContext
+        : TRequestContext
+        , std::enable_shared_from_this<TCheckContext>
+    {
+        TReadArg ReadArg;
+        ui32 NumPartsPending = 0;
+        std::unique_ptr<TEvBlobStorage::TEvCheckIntegrityResult> Result;
+
+        TCheckContext(TReadArg&& readArg)
+            : ReadArg(std::move(readArg))
+            , Result(new TEvBlobStorage::TEvCheckIntegrityResult(NKikimrProto::OK))
+        {}
+
+        void End(TQuery *query) {
+            query->OnCheckIntegrity(TCheckOutcome{std::move(Result)});
+        }
+    };
+
     bool TBlobDepotAgent::TQuery::IssueRead(TReadArg&& arg, TString& error) {
         STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA34, "IssueRead", (AgentId, Agent.LogId), (QueryId, GetQueryId()),
             (ReadId, arg.Tag), (Key, Agent.PrettyKey(arg.Key)), (Offset, arg.Offset), (Size, arg.Size),
@@ -277,4 +295,66 @@ namespace NKikimr::NBlobDepot {
         }
     }
 
+    void TBlobDepotAgent::TQuery::IssueCheckIntegrity(TReadArg&& arg) {
+        STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA62, "IssueCheckIntegrity", (AgentId, Agent.LogId),
+            (QueryId, GetQueryId()), (Key, Agent.PrettyKey(arg.Key)), (Value, arg.Value));
+
+        auto checkContext = std::make_shared<TCheckContext>(std::move(arg));
+
+        for (const auto& value : checkContext->ReadArg.Value.Chain) {
+            if (value.Blob) {
+                const auto& [blobId, groupId] = *value.Blob;
+                auto event = std::make_unique<TEvBlobStorage::TEvCheckIntegrity>(
+                    blobId, TInstant::Max(), checkContext->ReadArg.GetHandleClass);
+
+                STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA63, "issuing TEvCheckIntegrity", (AgentId, Agent.LogId), (QueryId, GetQueryId()),
+                    (Key, Agent.PrettyKey(checkContext->ReadArg.Key)), (GroupId, groupId), (Msg, *event));
+
+                Agent.SendToProxy(groupId, std::move(event), this, checkContext);
+                ++checkContext->NumPartsPending;
+            }
+        }
+    }
+
+    void TBlobDepotAgent::TQuery::HandleCheckIntegrityResult(const TRequestContext::TPtr& context,
+            TEvBlobStorage::TEvCheckIntegrityResult& msg) {
+        auto& checkContext = context->Obtain<TCheckContext>();
+
+        STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA64, "HandleCheckIntegrityResult", (AgentId, Agent.LogId),
+            (QueryId, GetQueryId()), (Key, Agent.PrettyKey(checkContext.ReadArg.Key)), (Msg, msg));
+
+        auto& result = checkContext.Result;
+        switch (msg.Status) {
+        case NKikimrProto::NODATA:
+            if (result->Status == NKikimrProto::OK) {
+                result->Status = NKikimrProto::NODATA;
+            }
+            break;
+        case NKikimrProto::ERROR:
+            result->Status = NKikimrProto::ERROR;
+            break;
+        default:
+            break;
+        }
+
+        if (msg.ErrorReason) {
+            if (result->ErrorReason) {
+                result->ErrorReason += "; ";
+            }
+            result->ErrorReason += msg.Id.ToString() + " " + msg.ErrorReason;
+        }
+
+        result->PlacementStatus = std::max(msg.PlacementStatus, result->PlacementStatus);
+        result->DataStatus = std::max(msg.DataStatus, result->DataStatus);
+
+        if (result->DataInfo) {
+            result->DataInfo += "; ";
+        }
+        result->DataInfo += msg.Id.ToString() + " " + msg.DataInfo;
+
+        if (!--checkContext.NumPartsPending) {
+            checkContext.End(this);
+        }
+    }
+
 } // NKikimr::NBlobDepot
diff --git a/ydb/core/blob_depot/agent/request.cpp b/ydb/core/blob_depot/agent/request.cpp
@@ -109,6 +109,7 @@ namespace NKikimr::NBlobDepot {
 
     template void TBlobDepotAgent::HandleOtherResponse(TEvBlobStorage::TEvGetResult::TPtr ev);
     template void TBlobDepotAgent::HandleOtherResponse(TEvBlobStorage::TEvPutResult::TPtr ev);
+    template void TBlobDepotAgent::HandleOtherResponse(TEvBlobStorage::TEvCheckIntegrityResult::TPtr ev);
 
     void TBlobDepotAgent::OnRequestComplete(ui64 id, TResponse response, TRequestsInFlight& map,
             std::shared_ptr<TEvBlobStorage::TExecutionRelay> executionRelay) {
diff --git a/ydb/core/blob_depot/agent/storage_check_integrity.cpp b/ydb/core/blob_depot/agent/storage_check_integrity.cpp
@@ -0,0 +1,93 @@
+#include "agent_impl.h"
+#include "blob_mapping_cache.h"
+
+namespace NKikimr::NBlobDepot {
+
+    template<>
+    TBlobDepotAgent::TQuery *TBlobDepotAgent::CreateQuery<TEvBlobStorage::EvCheckIntegrity>(
+            std::unique_ptr<IEventHandle> ev, TMonotonic received) {
+
+        class TCheckIntegrityQuery : public TBlobStorageQuery<TEvBlobStorage::TEvCheckIntegrity> {
+        public:
+            using TBlobStorageQuery::TBlobStorageQuery;
+
+            void Initiate() override {
+                if (IS_LOG_PRIORITY_ENABLED(NLog::PRI_TRACE, NKikimrServices::BLOB_DEPOT_EVENTS)) {
+                    BDEV_QUERY(BDEV25, "TEvCheckIntegrity_new", (U.BlobId, Request.Id));
+                }
+
+                TString blobId = Request.Id.AsBinaryString();
+
+                if (const TResolvedValue *value = Agent.BlobMappingCache.ResolveKey(blobId, this,
+                        std::make_shared<TRequestContext>(), false)) {
+                    ProcessResolveResult(value);
+                } else {
+                    STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA58, "resolve pending", (AgentId, Agent.LogId),
+                        (QueryId, GetQueryId()), (BlobId, Request.Id));
+                }
+            }
+
+            void ProcessResolveResult(const TKeyResolved& result) {
+                STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA59, "ProcessResolveResult", (AgentId, Agent.LogId),
+                    (QueryId, GetQueryId()), (Result, result));
+
+                if (result.Error()) {
+                    EndWithError(NKikimrProto::ERROR, result.GetErrorReason());
+                } else if (const TResolvedValue *value = result.GetResolvedValue(); !value) {
+                    EndWithError(NKikimrProto::NODATA, "no data");
+                } else {
+                    TReadArg arg{
+                        *value,
+                        Request.GetHandleClass,
+                        false,
+                        0,
+                        Request.Id.BlobSize(),
+                        0,
+                        std::nullopt,
+                        Request.Id.AsBinaryString(),
+                    };
+                    IssueCheckIntegrity(std::move(arg));
+                }
+            }
+
+            void OnCheckIntegrity(TCheckOutcome&& outcome) override {
+                STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA60, "OnCheckIntegrity", (AgentId, Agent.LogId), (QueryId, GetQueryId()),
+                    (Outcome, outcome));
+                TraceResponse(outcome.Result->Status);
+                TBlobStorageQuery::EndWithSuccess(std::move(outcome.Result));
+            }
+
+            void EndWithError(NKikimrProto::EReplyStatus status, const TString& errorReason) {
+                TraceResponse(status);
+                TBlobStorageQuery::EndWithError(status, errorReason);
+            }
+
+            void TraceResponse(NKikimrProto::EReplyStatus status) {
+                if (IS_LOG_PRIORITY_ENABLED(NLog::PRI_TRACE, NKikimrServices::BLOB_DEPOT_EVENTS)) {
+                    BDEV_QUERY(BDEV26, "TEvCheckIntegrity_end", (BlobId, Request.Id), (Status, status));
+                }
+            }
+
+            void ProcessResponse(ui64 /*id*/, TRequestContext::TPtr context, TResponse response) override {
+                if (auto *p = std::get_if<TKeyResolved>(&response)) {
+                    ProcessResolveResult(*p);
+                } else if (auto *p = std::get_if<TEvBlobStorage::TEvCheckIntegrityResult*>(&response)) {
+                    TQuery::HandleCheckIntegrityResult(context, **p);
+                } else if (std::holds_alternative<TTabletDisconnected>(response)) {
+                    STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA61, "TTabletDisconnected",
+                        (AgentId, Agent.LogId), (QueryId, GetQueryId()));
+                    EndWithError(NKikimrProto::ERROR, "Tablet disconnected");
+                } else {
+                    Y_ABORT();
+                }
+            }
+
+            ui64 GetTabletId() const override {
+                return Request.Id.TabletID();
+            }
+        };
+
+        return new TCheckIntegrityQuery(*this, std::move(ev), received);
+    }
+
+} // NKikimr::NBlobDepot
diff --git a/ydb/core/blob_depot/agent/ya.make b/ydb/core/blob_depot/agent/ya.make
@@ -36,6 +36,7 @@ LIBRARY()
         storage_get.cpp
         storage_get_block.cpp
         storage_block.cpp
+        storage_check_integrity.cpp
         storage_discover.cpp
         storage_range.cpp
         storage_collect_garbage.cpp
diff --git a/ydb/core/blobstorage/dsproxy/dsproxy_check_integrity_get.cpp b/ydb/core/blobstorage/dsproxy/dsproxy_check_integrity_get.cpp
@@ -188,8 +188,8 @@ class TBlobStorageGroupCheckIntegrityRequest : public TBlobStorageGroupRequestAc
             str << diskIdx << ": " << Info->CreateVDiskID(vDiskIdShort) << Endl;
         }
 
-        PendingResult->DataErrorInfo = str.Str();
-        PendingResult->DataErrorInfo += partsState.DataErrorInfo;
+        PendingResult->DataInfo = str.Str();
+        PendingResult->DataInfo += partsState.DataInfo;
 
         ReplyAndDie(NKikimrProto::OK);
     }
diff --git a/ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.h b/ydb/core/blobstorage/groupinfo/blobstorage_groupinfo.h
@@ -169,7 +169,7 @@ class TBlobStorageGroupInfo : public TThrRefBase {
 
         struct TPartsState {
             bool IsOk = true;
-            TString DataErrorInfo;
+            TString DataInfo;
         };
 
         virtual TPartsState GetDataState(const TLogoBlobID& id, const TPartsData& partsData) const = 0;
diff --git a/ydb/core/blobstorage/groupinfo/blobstorage_groupinfo_data_check.h b/ydb/core/blobstorage/groupinfo/blobstorage_groupinfo_data_check.h
@@ -88,7 +88,7 @@ class TDataIntegrityCheckerBlock42 : public TDataIntegrityCheckerBase {
             partsState.IsOk = false;
             layoutReport << "ERROR: There are unequal parts" << Endl;
         }
-        partsState.DataErrorInfo = layoutReport.Str();
+        partsState.DataInfo = layoutReport.Str();
 
         // checking erasure
         TStringStream erasureReport;
@@ -170,7 +170,7 @@ class TDataIntegrityCheckerBlock42 : public TDataIntegrityCheckerBase {
         if (!hasUnequalParts) {
             checkCombination();
             if (!erasureError) {
-                partsState.DataErrorInfo += erasureReport.Str();
+                partsState.DataInfo += erasureReport.Str();
                 return partsState;
             }
         }
@@ -202,7 +202,7 @@ class TDataIntegrityCheckerBlock42 : public TDataIntegrityCheckerBase {
             erasureReport << "ERROR: There are erasure restore fails" << Endl;
         }
 
-        partsState.DataErrorInfo += erasureReport.Str();
+        partsState.DataInfo += erasureReport.Str();
         return partsState;
     }
 };
@@ -267,7 +267,7 @@ class TDataIntegrityCheckerMirror : public TDataIntegrityCheckerBase {
             partsState.IsOk = false;
             layoutReport << "ERROR: There are unequal parts" << Endl;
         }
-        partsState.DataErrorInfo = layoutReport.Str();
+        partsState.DataInfo = layoutReport.Str();
 
         return partsState;
     }
diff --git a/ydb/core/blobstorage/ut_blobstorage/blob_depot.cpp b/ydb/core/blobstorage/ut_blobstorage/blob_depot.cpp
@@ -145,4 +145,13 @@ Y_UNIT_TEST_SUITE(BlobDepot) {
         TestVerifiedRandom(tenv, 8, 15, tenv.RegularGroups[0], 1000, 500, 10, { 10, 10, 3, 3, 2, 1, 1, 3, 3, 0 });
         // no blob depot restarts performed
     }
+
+    Y_UNIT_TEST(CheckIntegrity) {
+        ui32 seed;
+        LoadSeed(seed);
+        TBlobDepotTestEnvironment tenv(seed, 1, 8, TBlobStorageGroupType::Erasure4Plus2Block);
+
+        TestBasicCheckIntegrity(tenv, 1, tenv.RegularGroups[0]);
+        TestBasicCheckIntegrity(tenv, 1, tenv.BlobDepot);
+    }
 }
diff --git a/ydb/core/blobstorage/ut_blobstorage/blob_depot_event_managers.cpp b/ydb/core/blobstorage/ut_blobstorage/blob_depot_event_managers.cpp
diff --git a/ydb/core/blobstorage/ut_blobstorage/blob_depot_event_managers.h b/ydb/core/blobstorage/ut_blobstorage/blob_depot_event_managers.h
diff --git a/ydb/core/blobstorage/ut_blobstorage/blob_depot_test_functions.cpp b/ydb/core/blobstorage/ut_blobstorage/blob_depot_test_functions.cpp
diff --git a/ydb/core/blobstorage/ut_blobstorage/blob_depot_test_functions.h b/ydb/core/blobstorage/ut_blobstorage/blob_depot_test_functions.h
diff --git a/ydb/core/blobstorage/ut_blobstorage/check_integrity.cpp b/ydb/core/blobstorage/ut_blobstorage/check_integrity.cpp

Original file line number	Diff line number	Diff line change
`@@ -188,8 +188,8 @@ class TBlobStorageGroupCheckIntegrityRequest : public TBlobStorageGroupRequestAc`
`188`	`188`	`str << diskIdx << ": " << Info->CreateVDiskID(vDiskIdShort) << Endl;`
`189`	`189`	`}`
`190`	`190`
`191`		`- PendingResult->DataErrorInfo = str.Str();`
`192`		`- PendingResult->DataErrorInfo += partsState.DataErrorInfo;`
	`191`	`+ PendingResult->DataInfo = str.Str();`
	`192`	`+ PendingResult->DataInfo += partsState.DataInfo;`
`193`	`193`
`194`	`194`	`ReplyAndDie(NKikimrProto::OK);`
`195`	`195`	`}`
Original file line number	Diff line number	Diff line change
`@@ -88,7 +88,7 @@ class TDataIntegrityCheckerBlock42 : public TDataIntegrityCheckerBase {`
`88`	`88`	`partsState.IsOk = false;`
`89`	`89`	`layoutReport << "ERROR: There are unequal parts" << Endl;`
`90`	`90`	`}`
`91`		`- partsState.DataErrorInfo = layoutReport.Str();`
	`91`	`+ partsState.DataInfo = layoutReport.Str();`
`92`	`92`
`93`	`93`	`// checking erasure`
`94`	`94`	`TStringStream erasureReport;`
`@@ -170,7 +170,7 @@ class TDataIntegrityCheckerBlock42 : public TDataIntegrityCheckerBase {`
`170`	`170`	`if (!hasUnequalParts) {`
`171`	`171`	`checkCombination();`
`172`	`172`	`if (!erasureError) {`
`173`		`- partsState.DataErrorInfo += erasureReport.Str();`
	`173`	`+ partsState.DataInfo += erasureReport.Str();`
`174`	`174`	`return partsState;`
`175`	`175`	`}`
`176`	`176`	`}`
`@@ -202,7 +202,7 @@ class TDataIntegrityCheckerBlock42 : public TDataIntegrityCheckerBase {`
`202`	`202`	`erasureReport << "ERROR: There are erasure restore fails" << Endl;`
`203`	`203`	`}`
`204`	`204`
`205`		`- partsState.DataErrorInfo += erasureReport.Str();`
	`205`	`+ partsState.DataInfo += erasureReport.Str();`
`206`	`206`	`return partsState;`
`207`	`207`	`}`
`208`	`208`	`};`
`@@ -267,7 +267,7 @@ class TDataIntegrityCheckerMirror : public TDataIntegrityCheckerBase {`
`267`	`267`	`partsState.IsOk = false;`
`268`	`268`	`layoutReport << "ERROR: There are unequal parts" << Endl;`
`269`	`269`	`}`
`270`		`- partsState.DataErrorInfo = layoutReport.Str();`
	`270`	`+ partsState.DataInfo = layoutReport.Str();`
`271`	`271`
`272`	`272`	`return partsState;`
`273`	`273`	`}`