Put log chunks from killed owners on quarantine, if owner is still reading log (ydb-platform#12857)

Cherry-picked from eafe9a1

Author: SammyVimes

ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp

@@ -2129,16 +2129,20 @@ void TPDisk::KillOwner(TOwner owner, TOwnerRound killOwnerRound, TCompletionEven
         }
 
         TryTrimChunk(false, 0, NWilson::TSpan{});
+        bool readingLog = OwnerData[owner].ReadingLog();
         ui64 lastSeenLsn = 0;
         auto it = LogChunks.begin();
         while (it != LogChunks.end()) {
             if (it->OwnerLsnRange.size() > owner && it->OwnerLsnRange[owner].IsPresent) {
-                Y_ABORT_UNLESS(it->CurrentUserCount > 0);
-                it->CurrentUserCount--;
-                it->OwnerLsnRange[owner].IsPresent = false;
-                it->OwnerLsnRange[owner].FirstLsn = 0;
                 lastSeenLsn = Max(it->OwnerLsnRange[owner].LastLsn, lastSeenLsn);
-                it->OwnerLsnRange[owner].LastLsn = 0;
+
+                if (!readingLog) {
+                    Y_ABORT_UNLESS(it->CurrentUserCount > 0);
+                    it->CurrentUserCount--;
+                    it->OwnerLsnRange[owner].IsPresent = false;
+                    it->OwnerLsnRange[owner].FirstLsn = 0;
+                    it->OwnerLsnRange[owner].LastLsn = 0;
+                }
             }
             ++it;
         }
@@ -2376,22 +2380,50 @@ void TPDisk::ClearQuarantineChunks() {
         *Mon.QuarantineChunks = QuarantineChunks.size();
     }
 
+    bool haveChunksToRelease = false;
+
     {
         const auto it = std::partition(QuarantineOwners.begin(), QuarantineOwners.end(), [&] (TOwner i) {
             return Keeper.GetOwnerUsed(i) || OwnerData[i].HaveRequestsInFlight();
         });
         for (auto delIt = it; delIt != QuarantineOwners.end(); ++delIt) {
-            ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[*delIt].OperationLog, "Remove owner from quarantine, OwnerId# " << *delIt);
-            TOwnerRound ownerRound = OwnerData[*delIt].OwnerRound;
-            OwnerData[*delIt].Reset(false);
-            OwnerData[*delIt].OwnerRound = ownerRound;
-            Keeper.RemoveOwner(*delIt);
+            TOwner owner = *delIt;
+            ADD_RECORD_WITH_TIMESTAMP_TO_OPERATION_LOG(OwnerData[owner].OperationLog, "Remove owner from quarantine, OwnerId# " << owner);
+            TOwnerRound ownerRound = OwnerData[owner].OwnerRound;
+            OwnerData[owner].Reset(false);
+            OwnerData[owner].OwnerRound = ownerRound;
+            Keeper.RemoveOwner(owner);
+
+            ui64 lastSeenLsn = 0;
+            auto it = LogChunks.begin();
+            while (it != LogChunks.end()) {
+                if (it->OwnerLsnRange.size() > owner && it->OwnerLsnRange[owner].IsPresent) {
+                    Y_ABORT_UNLESS(it->CurrentUserCount > 0);
+                    ui32 userCount = --it->CurrentUserCount;
+                    it->OwnerLsnRange[owner].IsPresent = false;
+                    it->OwnerLsnRange[owner].FirstLsn = 0;
+                    lastSeenLsn = Max(it->OwnerLsnRange[owner].LastLsn, lastSeenLsn);
+                    it->OwnerLsnRange[owner].LastLsn = 0;
+
+                    if (userCount == 0) {
+                        haveChunksToRelease = true;
+                    }
+                }
+                ++it;
+            }
             LOG_NOTICE_S(*ActorSystem, NKikimrServices::BS_PDISK, "PDiskId# " << PDiskId
-                    << " removed ownerId# " << *delIt << " from chunks Keeper through QuarantineOwners");
+                    << " removed ownerId# " << (ui32)owner << " lastSeenLsn#" << lastSeenLsn << " from chunks Keeper through QuarantineOwners" << (haveChunksToRelease ? " along with log chunks" : ""));
         }
         QuarantineOwners.erase(it, QuarantineOwners.end());
         *Mon.QuarantineOwners = QuarantineOwners.size();
     }
+
+    if (haveChunksToRelease) {
+        THolder<TCompletionEventSender> completion(new TCompletionEventSender(this));
+        if (ReleaseUnusedLogChunks(completion.Get())) {
+            WriteSysLogRestorePoint(completion.Release(), TReqId(TReqId::KillOwnerSysLog, 0), {});
+        }
+    }
 }
 
 // Should be called to initiate TRIM (on chunk delete or prev trim done)

ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.h

@@ -142,7 +142,7 @@ class TPDisk : public IPDisk {
     ui64 InsaneLogChunks = 0;  // Set when pdisk sees insanely large log, to give vdisks a chance to cut it
     ui32 FirstLogChunkToParseCommits = 0;
 
-    // Chunks that is owned by killed owner, but has operations InFlight
+    // Chunks that are owned by killed owner, but have operations InFlight
     TVector<TChunkIdx> QuarantineChunks;
     TVector<TOwner> QuarantineOwners;
 

ydb/core/blobstorage/pdisk/blobstorage_pdisk_logreader.cpp

@@ -808,19 +808,41 @@ bool TLogReader::ProcessSectorSet(TSectorData *sector) {
                     << " LastGoodToWriteLogPosition# " << LastGoodToWriteLogPosition
                     << " Marker# LR018");
         } else {
-           Y_VERIFY_S(ChunkIdx == LogEndChunkIdx && SectorIdx >= LogEndSectorIdx, SelfInfo()
-                   << " File# " << __FILE__
-                   << " Line# " << __LINE__
-                   << " LogEndChunkIdx# " << LogEndChunkIdx
-                   << " LogEndSectorIdx# " << LogEndSectorIdx);
-            if (!(ChunkIdx == LogEndChunkIdx && SectorIdx >= LogEndSectorIdx)) {
-                LOG_WARN_S(*PDisk->ActorSystem, NKikimrServices::BS_PDISK, SelfInfo()
-                        << " In ProcessSectorSet got !restorator.GoodSectorFlags outside the LogEndSector."
+            bool outsideLogEnd = ChunkIdx == LogEndChunkIdx && SectorIdx >= LogEndSectorIdx;
+
+            if (!outsideLogEnd) {
+                // If read invalid data from the log (but not outside this owner's log bounds), check if the owner is on quarantine.
+                TGuard<TMutex> guard(PDisk->StateMutex);
+                TOwnerData &ownerData = PDisk->OwnerData[Owner];
+
+                if (ownerData.OnQuarantine) {
+                    LOG_WARN_S(*PDisk->ActorSystem, NKikimrServices::BS_PDISK, SelfInfo()
+                        << " In ProcessSectorSet got !restorator.GoodSectorFlags with owner on quarantine"
                         << " File# " << __FILE__
                         << " Line# " << __LINE__
                         << " LogEndChunkIdx# " << LogEndChunkIdx
                         << " LogEndSectorIdx# " << LogEndSectorIdx
-                        << " Marker# LR004");
+                        << " Marker# LR019");
+                    ReplyOk();
+                    return true;
+                }
+            }
+
+            Y_VERIFY_S(outsideLogEnd, SelfInfo()
+                    << " File# " << __FILE__
+                    << " Line# " << __LINE__
+                    << " LogEndChunkIdx# " << LogEndChunkIdx
+                    << " LogEndSectorIdx# " << LogEndSectorIdx);
+
+            if (outsideLogEnd) {
+                // It's ok.
+                LOG_WARN_S(*PDisk->ActorSystem, NKikimrServices::BS_PDISK, SelfInfo()
+                    << " In ProcessSectorSet got !restorator.GoodSectorFlags outside the LogEndSector"
+                    << " File# " << __FILE__
+                    << " Line# " << __LINE__
+                    << " LogEndChunkIdx# " << LogEndChunkIdx
+                    << " LogEndSectorIdx# " << LogEndSectorIdx
+                    << " Marker# LR004");
             }
         }
 

ydb/core/blobstorage/pdisk/blobstorage_pdisk_state.h

@@ -23,7 +23,7 @@ enum class EInitPhase {
 };
 
 enum EOwner {
-    OwnerSystem = 0, // Chunk0, SysLog chunks and CommonLog + just common log tracking, mens "for dynamic" in requests
+    OwnerSystem = 0, // Chunk0, SysLog chunks and CommonLog + just common log tracking, means "for dynamic" in requests
     OwnerUnallocated = 1, // Unallocated chunks, Trim scheduling, Slay commands
     OwnerBeginUser = 2,
     OwnerEndUser = 241,
@@ -156,6 +156,10 @@ struct TOwnerData {
         return LogReader || InFlight->ChunkWrites.load() || InFlight->ChunkReads.load() || InFlight->LogWrites.load();
     }
 
+    bool ReadingLog() const {
+        return bool(LogReader);
+    }
+
     TString ToString() const {
         TStringStream str;
         str << "TOwnerData {";

ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_env.h

@@ -26,6 +26,7 @@ struct TActorTestContext {
         ui32 ChunkSize = 128 * (1 << 20);
         bool SmallDisk = false;
         bool SuppressCompatibilityCheck = false;
+        TAutoPtr<TLogBackend> LogBackend = nullptr;
     };
 
 private:
@@ -71,7 +72,11 @@ struct TActorTestContext {
         IoContext = std::make_shared<NPDisk::TIoContextFactoryOSS>();
         appData->IoContextFactory = IoContext.get();
 
-        Runtime->SetLogBackend(IsLowVerbose ? CreateStderrBackend() : CreateNullBackend());
+        if (Settings.LogBackend) {
+            Runtime->SetLogBackend(Settings.LogBackend);
+        } else {
+            Runtime->SetLogBackend(IsLowVerbose ? CreateStderrBackend() : CreateNullBackend());
+        }
         Runtime->Initialize(TTestActorRuntime::TEgg{appData.Release(), nullptr, {}, {}});
         Runtime->SetLogPriority(NKikimrServices::BS_PDISK, NLog::PRI_NOTICE);
         Runtime->SetLogPriority(NKikimrServices::BS_PDISK_SYSLOG, NLog::PRI_NOTICE);
@@ -130,7 +135,7 @@ struct TActorTestContext {
         }
         return PDisk;
     }
-    
+
     void GracefulPDiskRestart(bool waitForRestart = true) {
         ui32 pdiskId = GetPDisk()->PDiskId;
 

ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut_races.cpp

@@ -254,11 +254,11 @@ Y_UNIT_TEST_SUITE(TPDiskRaces) {
     }
 
     Y_UNIT_TEST(KillOwnerWhileDecommittingWithInflight) {
-        TestKillOwnerWhileDecommitting(false, 20, 0, 10, 100);
+        TestKillOwnerWhileDecommitting(false, 20, 50, 10, 100);
     }
 
     Y_UNIT_TEST(KillOwnerWhileDecommittingWithInflightMock) {
-        TestKillOwnerWhileDecommitting(true, 20, 0, 10, 100);
+        TestKillOwnerWhileDecommitting(true, 20, 50, 10, 100);
     }
 
     void OwnerRecreationRaces(bool usePDiskMock, ui32 timeLimit, ui32 vdisksNum) {
@@ -325,6 +325,173 @@ Y_UNIT_TEST_SUITE(TPDiskRaces) {
     Y_UNIT_TEST(OwnerRecreationRaces) {
         OwnerRecreationRaces(false, 20, 1);
     }
+
+    void TestKillOwnerWhileReadingLog(ui32 timeLimit) {
+        // This test is not deterministic, so we run it multiple times to increase the chance of catching the bug.
+        // We expect to see quarantined log chunks in the log at least once (however locally it was seen every time).
+        // The original bug was crashing the server, so this test also tests this and that's why it doesn't break the cycle
+        // upon encountering quarantined log chunks.
+        bool capturedQuarantinedLogChunks = false;
+        THPTimer timer;
+        while (timer.Passed() < timeLimit) {
+            TStringStream ss;
+
+            TActorTestContext testCtx({ 
+                .IsBad = false,
+                .UsePDiskMock = false,
+                .LogBackend = new TStreamLogBackend(&ss),
+            });
+            const TString data = PrepareData(10_MB);
+
+            auto logNoTest = [&](TVDiskMock& mock, NPDisk::TCommitRecord rec) {
+                TString dataCopy = data;
+                auto evLog = MakeHolder<NPDisk::TEvLog>(mock.PDiskParams->Owner, mock.PDiskParams->OwnerRound, 0, TRcBuf(dataCopy),
+                        mock.GetLsnSeg(), nullptr);
+                evLog->Signature.SetCommitRecord();
+                evLog->CommitRecord = std::move(rec);
+                testCtx.Send(evLog.Release());
+            };
+
+            TVDiskMock mock(&testCtx);
+            mock.Init();
+
+            for (ui32 i = 0; i < 20; ++i) {
+                NPDisk::TCommitRecord rec;
+                logNoTest(mock, rec);
+                testCtx.Recv<NPDisk::TEvLogResult>();
+            }
+
+            testCtx.RestartPDiskSync();
+
+            mock.Init();
+
+            NPDisk::TLogPosition position{0, 0};
+
+            bool readCallbackCalled = false;
+
+            testCtx.TestCtx.SectorMap->SetReadCallback([&]() {
+                if (!readCallbackCalled) {
+                    readCallbackCalled = true;
+
+                    testCtx.Send(new NPDisk::TEvHarakiri(mock.PDiskParams->Owner, mock.PDiskParams->OwnerRound));
+                }
+            });
+
+            testCtx.Send(new NPDisk::TEvReadLog(mock.PDiskParams->Owner, mock.PDiskParams->OwnerRound, position));
+
+            testCtx.Recv<NPDisk::TEvHarakiriResult>();
+
+            {
+                TVDiskMock mock(&testCtx);
+                mock.Init();
+
+                for (ui32 i = 0; i < 13; ++i) {
+                    NPDisk::TCommitRecord rec;
+                    logNoTest(mock, rec);
+                    testCtx.Recv<NPDisk::TEvLogResult>();
+                }
+            }
+
+            if (!capturedQuarantinedLogChunks) {
+                TString log = ss.Str();
+                capturedQuarantinedLogChunks = log.Contains("along with log chunks");
+            }
+        }
+
+        UNIT_ASSERT(capturedQuarantinedLogChunks);
+    }
+
+    Y_UNIT_TEST(OwnerKilledWhileReadingLog) {
+        TestKillOwnerWhileReadingLog(20);
+    }
+
+    void TestKillOwnerWhileReadingLogAndThenKillLastOwner(ui32 timeLimit) {
+        bool capturedQuarantinedLogChunks = false;
+        THPTimer timer;
+        while (timer.Passed() < timeLimit) {
+            TStringStream ss;
+
+            TActorTestContext testCtx({ 
+                .IsBad = false,
+                .UsePDiskMock = false,
+                .LogBackend = new TStreamLogBackend(&ss),
+            });
+            const TString data = PrepareData(10_MB);
+
+            auto logNoTest = [&](TVDiskMock& mock, NPDisk::TCommitRecord rec) {
+                TString dataCopy = data;
+                auto evLog = MakeHolder<NPDisk::TEvLog>(mock.PDiskParams->Owner, mock.PDiskParams->OwnerRound, 0, TRcBuf(dataCopy),
+                        mock.GetLsnSeg(), nullptr);
+                evLog->Signature.SetCommitRecord();
+                evLog->CommitRecord = std::move(rec);
+                testCtx.Send(evLog.Release());
+            };
+
+            TVDiskMock mock1(&testCtx);
+            mock1.Init();
+
+            TVDiskMock mock2(&testCtx);
+            mock2.Init();
+
+            for (ui32 i = 0; i < 20; ++i) {
+                {
+                    NPDisk::TCommitRecord rec;
+                    logNoTest(mock1, rec);
+                    testCtx.Recv<NPDisk::TEvLogResult>();
+                }
+                {
+                    NPDisk::TCommitRecord rec;
+                    logNoTest(mock2, rec);
+                    testCtx.Recv<NPDisk::TEvLogResult>();
+                }
+            }
+
+            testCtx.RestartPDiskSync();
+
+            mock1.Init();
+            mock2.InitFull();
+
+            NPDisk::TLogPosition position{0, 0};
+
+            bool readCallbackCalled = false;
+
+            testCtx.TestCtx.SectorMap->SetReadCallback([&]() {
+                if (!readCallbackCalled) {
+                    readCallbackCalled = true;
+
+                    testCtx.Send(new NPDisk::TEvHarakiri(mock1.PDiskParams->Owner, mock1.PDiskParams->OwnerRound));
+                    testCtx.Send(new NPDisk::TEvHarakiri(mock2.PDiskParams->Owner, mock2.PDiskParams->OwnerRound));
+                }
+            });
+
+            testCtx.Send(new NPDisk::TEvReadLog(mock1.PDiskParams->Owner, mock1.PDiskParams->OwnerRound, position));
+
+            testCtx.Recv<NPDisk::TEvHarakiriResult>();
+            testCtx.Recv<NPDisk::TEvHarakiriResult>();
+            
+            {
+                TVDiskMock mock(&testCtx);
+                mock.Init();
+
+                for (ui32 i = 0; i < 30; ++i) {
+                    NPDisk::TCommitRecord rec;
+                    logNoTest(mock, rec);
+                    testCtx.Recv<NPDisk::TEvLogResult>();
+                }
+            }
+
+            if (!capturedQuarantinedLogChunks) {
+                TString log = ss.Str();
+                capturedQuarantinedLogChunks = log.Contains("along with log chunks");
+            }
+        }
+
+        UNIT_ASSERT(capturedQuarantinedLogChunks);
+    }
+
+    Y_UNIT_TEST(OwnerKilledWhileReadingLogAndThenKillLastOwner) {
+        TestKillOwnerWhileReadingLogAndThenKillLastOwner(20);
+    }
 }
 
 }