Add diagnostic information to TOwnerData,

Add diagnostic information to TOwnerData

9 files changed, 222 insertions, 17 deletions

diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_defs.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_defs.h
index 6c565d07a7d..7ec5566846f 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_defs.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_defs.h
@@ -104,4 +104,3 @@ template<>
 inline void Out<NKikimr::NPDisk::TOwner>(IOutputStream& os, const NKikimr::NPDisk::TPrintable_ui8& x) {
     os << static_cast<ui64>(x);
 }
-
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp
index fffa12c539c..6aed53c5b2a 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp
@@ -718,6 +718,7 @@ void TPDisk::AskVDisksToCutLogs(TOwner ownerFilter, bool doForce) {
                         ActorSystem->Send(new IEventHandle(OwnerData[chunkOwner].CutLogId, PDiskActor, cutLog.Release(),
                                     IEventHandle::FlagTrackDelivery, 0));
                         OwnerData[chunkOwner].AskedToCutLogAt = now;
+                        // ADD_OPERATION_TO_LOG(OwnerData[chunkOwner].OperationLog, "System owner asked to cut log at " << now);
                     } else {
                         LOG_INFO_S(*ActorSystem, NKikimrServices::BS_PDISK,
                                 "PDiskId# " << (ui32)PDiskId
@@ -761,6 +762,7 @@ void TPDisk::AskVDisksToCutLogs(TOwner ownerFilter, bool doForce) {
                     ActorSystem->Send(new IEventHandle(OwnerData[ownerFilter].CutLogId, PDiskActor, cutLog.Release(),
                                 IEventHandle::FlagTrackDelivery, 0));
                     OwnerData[ownerFilter].AskedToCutLogAt = now;
+                    // ADD_OPERATION_TO_LOG(OwnerData[ownerFilter].OperationLog, "User owner asked to cut log at " << now);
                 } else {
                     LOG_INFO_S(*ActorSystem, NKikimrServices::BS_PDISK,
                             "PDiskId# " << (ui32)PDiskId
@@ -1710,6 +1712,7 @@ bool TPDisk::YardInitForKnownVDisk(TYardInit &evYardInit, TOwner owner) {
     TVDiskID vDiskId = evYardInit.VDiskIdWOGeneration();
 
     TOwnerData &ownerData = OwnerData[owner];
+    ADD_OPERATION_TO_LOG(ownerData.OperationLog, "YardInitForKnownVDisk, evYardInit# " << evYardInit.ToString());
 
     ownerData.OwnerRound = evYardInit.OwnerRound;
     TOwnerRound ownerRound = evYardInit.OwnerRound;
@@ -1791,7 +1794,8 @@ bool TPDisk::YardInitStart(TYardInit &evYardInit) {
         }
         // TODO REPLY ERROR
         TOwnerData &data = OwnerData[owner];
-        Y_VERIFY_S(!data.HaveRequestsInFlight(), "owner# " << owner);
+        Y_VERIFY_S(!data.HaveRequestsInFlight(), "owner# " << owner <<
+            ", State: " << data.ToString() << ", Operation log: " << data.OperationLog.Print());
     }
     evYardInit.Owner = owner;
 
@@ -1808,6 +1812,7 @@ bool TPDisk::YardInitStart(TYardInit &evYardInit) {
     }
 
     // Update round and wait for all pending requests of old owner to finish
+    ADD_OPERATION_TO_LOG(ownerData.OperationLog, "YardInitStart, new OwnerRound# " << evYardInit.OwnerRound << ", State# " << ownerData.ToString());
     ownerData.OwnerRound = evYardInit.OwnerRound;
     return true;
 }
@@ -1830,22 +1835,23 @@ void TPDisk::YardInitFinish(TYardInit &evYardInit) {
         // TODO(cthulhu): don't allocate more owners than expected
         Keeper.AddOwner(owner, vDiskId);
 
-        OwnerData[owner].Reset(false);
-
+        TOwnerData& ownerData = OwnerData[owner];
+        ownerData.Reset(false);
         // A new owner is created.
 
         AtomicIncrement(TotalOwners);
-        OwnerData[owner].VDiskId = vDiskId;
+        ownerData.VDiskId = vDiskId;
         Y_VERIFY(SysLogFirstNoncesToKeep.FirstNonceToKeep[owner] <= SysLogRecord.Nonces.Value[NonceLog]);
         SysLogFirstNoncesToKeep.FirstNonceToKeep[owner] = SysLogRecord.Nonces.Value[NonceLog];
-        OwnerData[owner].CutLogId = evYardInit.CutLogId;
-        OwnerData[owner].WhiteboardProxyId = evYardInit.WhiteboardProxyId;
-        OwnerData[owner].VDiskSlotId = evYardInit.SlotId;
-        OwnerData[owner].OwnerRound = evYardInit.OwnerRound;
+        ownerData.CutLogId = evYardInit.CutLogId;
+        ownerData.WhiteboardProxyId = evYardInit.WhiteboardProxyId;
+        ownerData.VDiskSlotId = evYardInit.SlotId;
+        ownerData.OwnerRound = evYardInit.OwnerRound;
         VDiskOwners[vDiskId] = owner;
-        OwnerData[owner].Status = TOwnerData::VDISK_STATUS_SENT_INIT;
+        ownerData.Status = TOwnerData::VDISK_STATUS_SENT_INIT;
         SysLogRecord.OwnerVDisks[owner] = vDiskId;
-        ownerRound = OwnerData[owner].OwnerRound;
+        ownerRound = ownerData.OwnerRound;
+        ADD_OPERATION_TO_LOG(ownerData.OperationLog, "YardInitFinish, OwnerData# " << ownerData.ToString());
 
         AddCbsSet(owner);
 
@@ -1853,8 +1859,8 @@ void TPDisk::YardInitFinish(TYardInit &evYardInit) {
                 << " new owner is created. ownerId# " << owner
                 << " vDiskId# " << vDiskId.ToStringWOGeneration()
                 << " FirstNonceToKeep# " << SysLogFirstNoncesToKeep.FirstNonceToKeep[owner]
-                << " CutLogId# " << OwnerData[owner].CutLogId
-                << " ownerRound# " << OwnerData[owner].OwnerRound
+                << " CutLogId# " << ownerData.CutLogId
+                << " ownerRound# " << ownerData.OwnerRound
                 << " Marker# BPD02");
 
         AskVDisksToCutLogs(OwnerSystem, false);
@@ -1983,6 +1989,7 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
                 if (state.CommitState == TChunkState::DATA_ON_QUARANTINE) {
                     if (!pushedOwnerIntoQuarantine) {
                         pushedOwnerIntoQuarantine = true;
+                        ADD_OPERATION_TO_LOG(OwnerData[owner].OperationLog, "KillOwner(), Add owner to quarantine, DATA_ON_QUARANTINE, OwnerId# " << (ui32)owner);
                         QuarantineOwners.push_back(owner);
                         LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                                 << " push ownerId# " << owner
@@ -2018,6 +2025,7 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
 
                     if (!pushedOwnerIntoQuarantine) {
                         pushedOwnerIntoQuarantine = true;
+                        ADD_OPERATION_TO_LOG(OwnerData[owner].OperationLog, "KillOwner(), Add owner to quarantine, OwnerId# " << (ui32)owner);
                         QuarantineOwners.push_back(owner);
                         LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                                 << " push ownerId# " << owner << " into quarantine");
@@ -2028,6 +2036,7 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
             }
         }
         if (!pushedOwnerIntoQuarantine) {
+            ADD_OPERATION_TO_LOG(OwnerData[owner].OperationLog, "KillOwner(), Remove owner without quarantine, OwnerId# " << (ui32)owner);
             Keeper.RemoveOwner(owner);
             LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                     << " removed ownerId# " << owner << " from chunks Keeper");
@@ -2280,6 +2289,7 @@ void TPDisk::ClearQuarantineChunks() {
             return Keeper.GetOwnerUsed(i);
         });
         for (auto delIt = it; delIt != QuarantineOwners.end(); ++delIt) {
+            ADD_OPERATION_TO_LOG(OwnerData[*delIt].OperationLog, "Remove owner from quarantine, OwnerId# " << (ui32)(*delIt));
             Keeper.RemoveOwner(*delIt);
             LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                     << " removed ownerId# " << *delIt << " from chunks Keeper through QuarantineOwners");
@@ -2826,6 +2836,7 @@ bool TPDisk::PreprocessRequest(TRequestBase *request) {
                     new TEvChunkWriteResult(NKikimrProto::OK, ev.ChunkIdx, ev.Cookie,
                         GetStatusFlags(ev.Owner, ev.OwnerGroupType), TString()));
 
+            // ADD_OPERATION_TO_LOG(ownerData.OperationLog, "Preprocessing chunk write request, write size# " << ev.TotalSize);
             ++state.OperationsInProgress;
             ++ownerData.InFlight->ChunkWrites;
             auto onDestroy = [&, inFlight = ownerData.InFlight]() {
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_http.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_http.cpp
index 0df4d828ed4..102ed626fe7 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_http.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_http.cpp
@@ -202,6 +202,7 @@ void TPDisk::OutputHtmlOwners(TStringStream &str) {
                     TABLEH() { str << "FirstNonceToKeep"; }
                     TABLEH() { str << "AskedToCutLogAt"; }
                     TABLEH() { str << "CutLogAt"; }
+                    TABLEH() { str << "OperationLog"; }
                 }
             }
             TABLEBODY() {
@@ -226,6 +227,19 @@ void TPDisk::OutputHtmlOwners(TStringStream &str) {
                                     str << data.CutLogAt;
                                 }
                             }
+                            TABLED() {
+                                str << "<button type='button' class='btn btn-default' data-toggle='collapse' style='margin:5px' \
+                                    data-target='#operationLogCollapse" << owner << 
+                                    "'>Show last " << OwnerData[owner].OperationLog.Size << " operations</button>";
+
+                                str << "<div id='operationLogCollapse" << owner << "' class='collapse'>";
+                                for (ui32 i = OwnerData[owner].OperationLog.RecordIdx, ctr = 0; 
+                                        ctr < OwnerData[owner].OperationLog.Size; 
+                                        ++ctr, i = (i == 0) ? OwnerData[owner].OperationLog.Size - 1 : i - 1) {
+                                    str << OwnerData[owner].OperationLog.Records[i] << "<br>"; 
+                                }
+                                str << "</div>";
+                            }
                         }
                     }
                 }
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_log.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_log.cpp
index 3ce5c470e0f..3dd40d6864f 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_log.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_log.cpp
@@ -217,6 +217,7 @@ void TPDisk::ProcessChunk0(const NPDisk::TEvReadLogResult &readLogResult) {
         id.GroupGeneration = -1;  // Clear GroupGeneration in sys log record (for compatibility)
         OwnerData[i].VDiskId = id;
         OwnerData[i].Status = TOwnerData::VDISK_STATUS_HASNT_COME;
+        ADD_OPERATION_TO_LOG(OwnerData[i].OperationLog, "Processing Chunk0, vdisk id# " << id.ToString());
         if (id != TVDiskID::InvalidId) {
             VDiskOwners[id] = TOwner(i);
             AtomicIncrement(TotalOwners);
@@ -433,6 +434,7 @@ void TPDisk::ProcessLogReadQueue() {
         {
             TLogRead &logRead = *static_cast<TLogRead*>(req);
             auto& ownerData = OwnerData[logRead.Owner];
+
             ownerData.Status = TOwnerData::VDISK_STATUS_READING_LOG;
             TLogPosition logStartPosition{0, 0};
             if (logRead.Owner < OwnerData.size() && ownerData.VDiskId != TVDiskID::InvalidId) {
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_logreader.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_logreader.cpp
index 05a22e0a537..b5c14971905 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_logreader.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_logreader.cpp
@@ -165,6 +165,7 @@ void TPDisk::ProcessReadLogRecord(TLogRecordHeader &header, TString &data, NPDis
             {
                 TGuard<TMutex> guard(StateMutex);
                 TOwnerData &ownerData = OwnerData[header.OwnerId];
+
                 if (ownerData.VDiskId != TVDiskID::InvalidId) {
                     if (!ownerData.IsNextLsnOk(header.OwnerLsn)) {
                         TStringStream str;
@@ -175,7 +176,7 @@ void TPDisk::ProcessReadLogRecord(TLogRecordHeader &header, TString &data, NPDis
                             << " header.OwnerLsn# " << header.OwnerLsn
                             << " nonce# " << nonce
                             << Endl;
-                        Y_FAIL_S(str.Str());
+                        Y_FAIL_S(str.Str() << " operation log# " << ownerData.OperationLog.Print());
                     }
                     ownerData.LastSeenLsn = header.OwnerLsn;
                 }
@@ -957,6 +958,7 @@ void TLogReader::ReplyOk() {
             }
             // End of log reached
             if (OwnerVDiskId != TVDiskID::InvalidId) {
+                ADD_OPERATION_TO_LOG(ownerData.OperationLog, "Has read the whole log");
                 ownerData.HasReadTheWholeLog = true;
             }
         }
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h
index 7e981933fbb..0099c9eb36b 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h
@@ -141,6 +141,16 @@ public:
         v.GroupGeneration = -1;
         return v;
     }
+
+    TString ToString() const {
+        TStringStream str;
+        str << "TYardInit {";
+        str << "VDisk# " << VDisk.ToString();
+        str << " PDiskGuid# " << PDiskGuid;
+        str << " SlotId# " << SlotId;
+        str << "}";
+        return str.Str();
+    }
 };
 
 //
@@ -311,6 +321,18 @@ public:
     void SetOnDestroy(std::function<void()> onDestroy) {
         OnDestroy = std::move(onDestroy);
     }
+
+    TString ToString() const {
+        TStringStream str;
+        str << "TLogWrite {";
+        str << "EstimatedChunkIdx# " << EstimatedChunkIdx;
+        str << " LsnSegmentStart# " << LsnSegmentStart;
+        str << " Lsn# " << Lsn;
+        str << " Result# " << (!Result ? "is empty" : Result->ToString());
+        str << " OnDestroy is " << (!OnDestroy ? "not " : "") << "set";
+        str << "}";
+        return str.Str();
+    }
 };
 
 class TCompletionChunkRead;
@@ -911,6 +933,18 @@ public:
     ERequestType GetType() const override {
         return ERequestType::RequestLogCommitDone;
     }
+
+    TString ToString() const {
+        TStringStream str;
+        str << "TLogCommitDone {";
+        str << "OwnerId# " << (ui32)OwnerId;
+        str << " OwnerRound# " << OwnerRound;
+        str << " Lsn# " << Lsn;
+        str << " CommitedChunks.size()# " << CommitedChunks.size();
+        str << " DeletedChunks.size()# " << DeletedChunks.size();
+        str << "}";
+        return str.Str();
+    }
 };
 
 //
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h
index 66d6c384360..87284fde948 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h
@@ -3,6 +3,7 @@
 
 #include "blobstorage_pdisk.h"
 #include "blobstorage_pdisk_logreader_base.h"
+#include "blobstorage_pdisk_tools.h"
 
 #include <ydb/core/util/metrics.h>
 
@@ -80,6 +81,8 @@ struct TOwnerData {
 
     bool OnQuarantine = false;
 
+    TOperationLog<8> OperationLog;
+
     TOwnerData()
       : InFlight(new TOwnerInflight)
     {}
@@ -139,11 +142,65 @@ struct TOwnerData {
     }
 
     bool HaveRequestsInFlight() const {
-        return LogReader || InFlight->ChunkWrites || InFlight->ChunkReads || InFlight->LogWrites;
+        return LogReader || InFlight->ChunkWrites.load() || InFlight->ChunkReads.load() || InFlight->LogWrites.load();
+    }
+
+    TString ToString() const {
+        TStringStream str;
+        str << "TOwnerData {";
+        str << "VDiskId# " << VDiskId.ToString();
+        str << " Status# " << RenderStatus(Status);
+        str << " CurrentFirstLsnToKeep# " << CurrentFirstLsnToKeep;
+        str << " LastWrittenCommitLsn# " << LastWrittenCommitLsn;
+        str << " LogRecordsInitiallyRead# " << LogRecordsInitiallyRead;
+        str << " LogRecordsConsequentlyRead# " << LogRecordsConsequentlyRead;
+        str << " OwnerRound# " << OwnerRound;
+        str << " AskedToCutLogAt# " << AskedToCutLogAt;
+        str << " CutLogAt# " << CutLogAt;
+        str << " LastSeenLsn# " << LastSeenLsn;
+        if (HasAlreadyLoggedThisIncarnation) {
+            str << " HasAlreadyLoggedThisIncarnation";
+        }
+        if (HasReadTheWholeLog) {
+            str << " HasReadTheWholeLog";
+        }
+        str << " VDiskSlotId# " << VDiskSlotId;
+        if (InFlight) {
+            str << " Inflight {";
+            str << " ChunkWrites# " << InFlight->ChunkWrites.load();
+            str << " ChunkReads# " << InFlight->ChunkReads.load();
+            str << " LogWrites# " << InFlight->LogWrites.load();
+            str << " }";
+        }
+        str << "}";
+
+        return str.Str();
     }
 
     void Reset(bool quarantine = false) {
-        *this = TOwnerData{};
+        StartingPoints.clear();
+        VDiskId = TVDiskID::InvalidId;
+        Status = EVDiskStatus::VDISK_STATUS_DEFAULT;
+        CurrentFirstLsnToKeep = 0;
+        LastWrittenCommitLsn = 0;
+        CutLogId = TActorId();
+        WhiteboardProxyId = TActorId();
+        LogRecordsInitiallyRead = 0;
+        LogRecordsConsequentlyRead = 0;
+        OwnerRound = 0;
+        AskedToCutLogAt = TInstant();
+        CutLogAt = TInstant();
+        LastSeenLsn = 0;
+        HasAlreadyLoggedThisIncarnation = false;
+        HasReadTheWholeLog = false;
+        LogStartPosition = TLogPosition{0, 0};
+        ReadThroughput = NMetrics::TDecayingAverageValue<ui64, NMetrics::DurationPerMinute, NMetrics::DurationPerSecond>();
+        WriteThroughput = NMetrics::TDecayingAverageValue<ui64, NMetrics::DurationPerMinute, NMetrics::DurationPerSecond>();
+        VDiskSlotId = 0;
+
+        LogReader.Reset();
+        InFlight.Reset(TIntrusivePtr<TOwnerInflight>(new TOwnerInflight));
+
         OnQuarantine = quarantine;
     }
 };
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h
index 47021792407..221bccb7733 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h
@@ -50,6 +50,29 @@ bool ReadPDiskFormatInfo(const TString &path, const NPDisk::TMainKey &mainKey, T
     const bool doLock = false, TIntrusivePtr<NPDisk::TSectorMap> sectorMap = nullptr);
 
 
-} // NKikimr
+// Size is better to be 2^n, to optimize mod operation
+template <ui32 S>
+struct TOperationLog {
+    constexpr static ui32 Size = S;
+    std::array<TString, S> Records;
+    std::atomic<ui64> RecordIdx = std::atomic<ui64>(0);
+
+    TString Print() const {
+        TStringStream str;
+        str << "[ " ;
+        /* Print OperationLog records from newer to older */ 
+        for (ui32 i = RecordIdx.load(), ctr = 0; ctr < Size; ++ctr, i = (i == 0) ? Size - 1 : i - 1) {
+            str << "Record# " << ctr <<  ": { " << Records[i] << " }, ";
+        }
+        str << " ]";
+        return str.Str();
+    }
+};
 
+} // NKikimr
 
+#define ADD_OPERATION_TO_LOG(log, record)                                                               \
+    do {                                                                                                \
+        ui32 idx = log.RecordIdx.fetch_add(1);                                                          \
+        log.Records[idx % log.Size] = TStringBuilder() << TInstant::Now().ToString() << " " << record;  \
+    } while (false)
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp
index 9b6fdf5090a..bb1e680b192 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp
@@ -8,6 +8,7 @@
 #include <ydb/core/testlib/actors/test_runtime.h>
 
 #include <util/system/hp_timer.h>
+#include <util/random/random.h>
 
 namespace NKikimr {
 
@@ -259,6 +260,68 @@ Y_UNIT_TEST_SUITE(TPDiskRaces) {
     Y_UNIT_TEST(KillOwnerWhileDecommittingWithInflightMock) {
         TestKillOwnerWhileDecommitting(true, 20, 0, 10, 100);
     }
+
+    void OwnerRecreationRaces(bool usePDiskMock, ui32 timeLimit, ui32 vdisksNum) {
+        TActorTestContext testCtx(false, usePDiskMock);
+        const TString data = PrepareData(8);
+
+        std::vector<TVDiskMock> mocks;
+        enum EMockState {
+            Empty,
+            InitStarted, 
+            InitFinished,
+            KillStarted,
+            KillFinished
+        };
+        std::vector<EMockState> mockState(vdisksNum, EMockState::Empty);
+        for (ui32 i = 0; i < vdisksNum; ++i) {
+            mocks.push_back(TVDiskMock(&testCtx));
+        }
+
+        THPTimer timer;
+        while (timer.Passed() < timeLimit) {
+            ui32 i = RandomNumber(vdisksNum);
+            ui32 action = RandomNumber<ui32>(10);
+            if (action != 0 && mocks[i].PDiskParams) {
+                auto evLog = MakeHolder<NPDisk::TEvLog>(mocks[i].PDiskParams->Owner, mocks[i].PDiskParams->OwnerRound, 0, TContiguousData(data),
+                        mocks[i].GetLsnSeg(), nullptr);
+                evLog->Signature = TLogSignature::SignatureLogoBlobOpt;
+                testCtx.Send(evLog.Release());
+            } else {
+                switch (mockState[i]) {
+                case EMockState::InitStarted:
+                    {
+                        auto res = testCtx.Recv<NPDisk::TEvYardInitResult>();
+                        UNIT_ASSERT_VALUES_EQUAL(res->Status, NKikimrProto::OK);
+                        mocks[i].PDiskParams = res->PDiskParams;
+                        mocks[i].OwnerRound = res->PDiskParams->OwnerRound;
+                    }
+                    break;
+                case EMockState::InitFinished:
+                    {
+                        auto evSlay = MakeHolder<NPDisk::TEvSlay>(mocks[i].VDiskID, mocks[i].OwnerRound++, 0, 0);
+                        testCtx.Send(evSlay.Release());
+                    }
+                    break;
+                case EMockState::KillStarted:
+                    {
+                        auto res = testCtx.Recv<NPDisk::TEvSlayResult>();
+                    }
+                    break;
+                case EMockState::Empty: case EMockState::KillFinished:
+                    {
+                        auto evInit = MakeHolder<NPDisk::TEvYardInit>(mocks[i].OwnerRound++, mocks[i].VDiskID, testCtx.TestCtx.PDiskGuid);
+                        testCtx.Send(evInit.Release());
+                    }
+                    break;
+                }
+            }
+        }
+    }
+
+    Y_UNIT_TEST(OwnerRecreationRaces) {
+        OwnerRecreationRaces(false, 20, 1);
+    }
 }
 
 }