Add TOperationLog to track owner updates in PDisk,

Fix mailbox overflow

10 files changed, 320 insertions, 18 deletions

diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_defs.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_defs.h
index 6c565d07a7..7ec5566846 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 3884ae9a69..7e804978e3 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp
@@ -737,6 +737,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_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[chunkOwner].OperationLog, "System owner asked to cut log, OwnerId# " << chunkOwner);
                     } else {
                         LOG_INFO_S(*ActorSystem, NKikimrServices::BS_PDISK,
                                 "PDiskId# " << (ui32)PDiskId
@@ -780,6 +781,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_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[ownerFilter].OperationLog, "User owner asked to cut log, OwnerId# " << ownerFilter);
                 } else {
                     LOG_INFO_S(*ActorSystem, NKikimrServices::BS_PDISK,
                             "PDiskId# " << (ui32)PDiskId
@@ -1735,6 +1737,8 @@ bool TPDisk::YardInitForKnownVDisk(TYardInit &evYardInit, TOwner owner) {
     TVDiskID vDiskId = evYardInit.VDiskIdWOGeneration();
 
     TOwnerData &ownerData = OwnerData[owner];
+    ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(ownerData.OperationLog, "YardInitForKnownVDisk, OwnerId# " << owner
+            << ", evYardInit# " << evYardInit.ToString());
 
     ownerData.OwnerRound = evYardInit.OwnerRound;
     TOwnerRound ownerRound = evYardInit.OwnerRound;
@@ -1833,6 +1837,8 @@ bool TPDisk::YardInitStart(TYardInit &evYardInit) {
     }
 
     // Update round and wait for all pending requests of old owner to finish
+    ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(ownerData.OperationLog, "YardInitStart, OwnerId# "
+            << owner << ", new OwnerRound# " << evYardInit.OwnerRound);
     ownerData.OwnerRound = evYardInit.OwnerRound;
     return true;
 }
@@ -1855,22 +1861,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_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(ownerData.OperationLog, "YardInitFinish, OwnerId# " << owner);
 
         AddCbsSet(owner);
 
@@ -1878,8 +1885,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);
@@ -2010,6 +2017,8 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
                 if (state.CommitState == TChunkState::DATA_ON_QUARANTINE) {
                     if (!pushedOwnerIntoQuarantine) {
                         pushedOwnerIntoQuarantine = true;
+                        ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[owner].OperationLog, "KillOwner(), Add owner to quarantine, " 
+                                << "CommitState# DATA_ON_QUARANTINE, OwnerId# " << owner);
                         QuarantineOwners.push_back(owner);
                         LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                                 << " push ownerId# " << owner
@@ -2045,6 +2054,7 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
 
                     if (!pushedOwnerIntoQuarantine) {
                         pushedOwnerIntoQuarantine = true;
+                        ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[owner].OperationLog, "KillOwner(), Add owner to quarantine, OwnerId# " << owner);
                         QuarantineOwners.push_back(owner);
                         LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                                 << " push ownerId# " << owner << " into quarantine");
@@ -2055,6 +2065,7 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
             }
         }
         if (!pushedOwnerIntoQuarantine) {
+            ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[owner].OperationLog, "KillOwner(), Remove owner without quarantine, OwnerId# " << owner);
             Keeper.RemoveOwner(owner);
             LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                     << " removed ownerId# " << owner << " from chunks Keeper");
@@ -2307,6 +2318,7 @@ void TPDisk::ClearQuarantineChunks() {
             return Keeper.GetOwnerUsed(i);
         });
         for (auto delIt = it; delIt != QuarantineOwners.end(); ++delIt) {
+            ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[*delIt].OperationLog, "Remove owner from quarantine, OwnerId# " << *delIt);
             Keeper.RemoveOwner(*delIt);
             LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
                     << " removed ownerId# " << *delIt << " from chunks Keeper through QuarantineOwners");
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_http.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl_http.cpp
index 0df4d828ed..1ad0cf2743 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,17 @@ 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 = 0; i < OwnerData[owner].OperationLog.Size(); ++i) {
+                                    str << OwnerData[owner].OperationLog[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 43e49b5592..95fe46107c 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_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[i].OperationLog, "Processing Chunk0, OwnerId# " << i);
         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 841b89a756..020452beec 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.ToString());
                     }
                     ownerData.LastSeenLsn = header.OwnerLsn;
                 }
@@ -957,6 +958,7 @@ void TLogReader::ReplyOk() {
             }
             // End of log reached
             if (OwnerVDiskId != TVDiskID::InvalidId) {
+                ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(ownerData.OperationLog, "Has read the whole log, OwnerId# " << Owner);
                 ownerData.HasReadTheWholeLog = true;
             }
         }
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_requestimpl.h
index e9a731d46a..a819fae0d8 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 66d6c38436..9312b4167a 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h
@@ -3,8 +3,10 @@
 
 #include "blobstorage_pdisk.h"
 #include "blobstorage_pdisk_logreader_base.h"
+#include "blobstorage_pdisk_tools.h"
 
 #include <ydb/core/util/metrics.h>
+#include <ydb/core/debug_tools/operation_log.h>
 
 namespace NKikimr {
 namespace NPDisk {
@@ -80,6 +82,8 @@ struct TOwnerData {
 
     bool OnQuarantine = false;
 
+    TOperationLog<8> OperationLog;
+
     TOwnerData()
       : InFlight(new TOwnerInflight)
     {}
@@ -139,11 +143,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 8a44c6a6bb..b3150e96ae 100644
--- a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h
+++ b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_tools.h
@@ -49,7 +49,4 @@ void FormatPDisk(TString path, ui64 diskSizeBytes, ui32 sectorSizeBytes, ui32 us
 bool ReadPDiskFormatInfo(const TString &path, const NPDisk::TMainKey &mainKey, TPDiskInfo &outInfo,
     const bool doLock = false, TIntrusivePtr<NPDisk::TSectorMap> sectorMap = nullptr);
 
-
 } // NKikimr
-
-
diff --git a/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp b/ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp
index eb80d371ee..169802a2ee 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,71 @@ 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 });
+
+        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, TRcBuf(PrepareData(1)),
+                        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.Reset(res->PDiskParams);
+                        mocks[i].OwnerRound = res->PDiskParams->OwnerRound;
+                        mockState[i] = EMockState::InitFinished;
+                    }
+                    break;
+                case EMockState::InitFinished:
+                    {
+                        auto evSlay = MakeHolder<NPDisk::TEvSlay>(mocks[i].VDiskID, mocks[i].OwnerRound++, 0, 0);
+                        testCtx.Send(evSlay.Release());
+                        mockState[i] = EMockState::KillStarted;
+                    }
+                    break;
+                case EMockState::KillStarted:
+                    {
+                        testCtx.Recv<NPDisk::TEvSlayResult>();
+                        mockState[i] = EMockState::KillFinished;
+                    }
+                    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());
+                        mockState[i] = EMockState::InitStarted;
+                    }
+                    break;
+                }
+            }
+        }
+    }
+
+    Y_UNIT_TEST(OwnerRecreationRaces) {
+        OwnerRecreationRaces(false, 20, 1);
+    }
 }
 
 }
diff --git a/ydb/core/debug_tools/operation_log.h b/ydb/core/debug_tools/operation_log.h
new file mode 100644
index 0000000000..3e44e6c92b
--- /dev/null
+++ b/ydb/core/debug_tools/operation_log.h
@@ -0,0 +1,120 @@
+#pragma once
+#include <util/string/builder.h>
+#include <util/system/mutex.h>
+#include <util/system/types.h>
+#include <util/generic/utility.h>
+
+#include <array>
+
+namespace NKikimr {
+
+template <ui32 S>
+class TOperationLog {
+private:
+    class const_iterator {
+        friend class TOperationLog;
+    public:
+        const_iterator(ui64 position, const TOperationLog& container)
+            : Container(container)
+            , Position(position) {
+        }
+
+        bool operator==(const const_iterator& other) const {
+            return other.Position == Position;
+        }
+
+        bool operator!=(const const_iterator& other) const {
+            return other.Position != Position;
+        }
+
+        const_iterator& operator++() {
+            Position--;
+            return *this;
+        }
+
+        const_iterator& operator++(int) {
+            Position--;
+            return *this;
+        }
+
+        const TString& operator*() const {
+            const TString& str = Container.GetByPosition(Position);
+            return str;
+        }
+
+    private:
+        ui64 GetPosition() {
+            return Position;
+        }
+
+        const TOperationLog& Container;
+        ui64 Position;
+    };
+
+    friend class const_iterator;
+
+private:
+    constexpr static ui32 Capacity = S;
+    std::array<TString, S> Records;
+    // RecordIdx is shifted to prevent underflow on decrement
+    std::atomic<ui64> NextRecordPosition = Capacity;
+
+public:
+    TOperationLog() = default;
+
+    TOperationLog(const TOperationLog& other) = delete;
+    TOperationLog& operator=(const TOperationLog& other) = delete;
+
+    TOperationLog(TOperationLog&& other) = default;
+    TOperationLog& operator=(TOperationLog&& other) = default;
+
+    const_iterator begin() const {
+        return const_iterator(NextRecordPosition.load() - 1, *this);
+    }
+
+    const_iterator end() const {
+        return const_iterator(NextRecordPosition.load() - 1 - Size(), *this);
+    }
+
+    const TString& GetByPosition(ui64 position) const {
+        return Records[position % Capacity];
+    }
+
+    const TString& operator[](ui32 idx) const {
+        Y_VERIFY(idx < Size());
+        return Records[(NextRecordPosition.load() - 1 - idx) % Capacity];
+    }
+
+    ui32 Size() const {
+        return Min(NextRecordPosition.load() - Capacity, static_cast<ui64>(Capacity));
+    }
+
+    void AddRecord(const TString& value) {
+        Records[NextRecordPosition.fetch_add(1) % Capacity] = value;
+    }
+
+    TString ToString() const {
+        TStringStream str;
+        str << "[ " ;
+        /* Print OperationLog records from newer to older */ 
+        ui32 ctr = 0;
+        for (auto it = begin(); it != end(); ++it, ++ctr) {
+            str << "Record# " << ctr <<  " : { " << GetByPosition(it.GetPosition()) << " }, ";
+        }
+        str << " ]";
+        return str.Str();
+    }
+};
+
+} // NKikimr
+
+#define ADD_RECORD_TO_OPERATION_LOG(log, record)    \
+    do {                                            \
+        log.AddRecord(TStringBuilder() << record);  \
+    } while (false)
+
+
+#define ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(log, record)                         \
+    do {                                                                                \
+        log.AddRecord(TStringBuilder() << TInstant::Now().ToString() << " " << record); \
+    } while (false)