Fixes to maintenance: purge ledger headers in the background, properly do SCP history maintenance, improve tests

Author: marta-lokhova

src/history/test/HistoryTests.cpp

@@ -6,6 +6,7 @@
 #include "bucket/test/BucketTestUtils.h"
 #include "catchup/LedgerApplyManagerImpl.h"
 #include "catchup/test/CatchupWorkTests.h"
+#include "herder/HerderPersistence.h"
 #include "history/CheckpointBuilder.h"
 #include "history/FileTransferInfo.h"
 #include "history/HistoryArchiveManager.h"
@@ -15,6 +16,7 @@
 #include "historywork/GunzipFileWork.h"
 #include "historywork/GzipFileWork.h"
 #include "historywork/PutHistoryArchiveStateWork.h"
+#include "ledger/LedgerHeaderUtils.h"
 #include "ledger/LedgerManager.h"
 #include "main/Maintainer.h"
 #include "main/PersistentState.h"
@@ -1440,12 +1442,29 @@ TEST_CASE("persist publish queue", "[history][publish][acceptance]")
             // Trim history after publishing whenever possible.
             app1->getMaintainer().performMaintenance(50000);
         }
+
+        // Verify old history got trimmed
+        XDROutputFileStream out(app1->getClock().getIOContext(), true);
+        // Ledgers to add to genesis, maintenance keeps at least one checkpoint
+        // of data, so last ledger trimmed is (5 * 8 - 1) - 8 = 31
+        uint32_t publishedCount = 30;
+        auto scp = app1->getHerderPersistence().copySCPHistoryToStream(
+            app1->getDatabase().getRawSession(),
+            LedgerManager::GENESIS_LEDGER_SEQ, publishedCount, out);
+        REQUIRE(scp == 0);
+
+        CheckpointBuilder cb(*app1);
+        auto headers = LedgerHeaderUtils::copyToStream(
+            app1->getDatabase().getRawSession(),
+            LedgerManager::GENESIS_LEDGER_SEQ, publishedCount, cb);
+        REQUIRE(headers == 0);
+
         // We should have either an empty publish queue or a
-        // ledger sometime after the 5th checkpoint
+        // ledger sometime after the 5th checkpoint (ledger 5 * 8 - 1 = 39)
         auto minLedger =
             HistoryManager::getMinLedgerQueuedToPublish(hm1.getConfig());
         LOG_INFO(DEFAULT_LOG, "minLedger {}", minLedger);
-        bool okQueue = minLedger == 0 || minLedger >= 35;
+        bool okQueue = minLedger == 0 || minLedger >= 39;
         REQUIRE(okQueue);
     }
 }

src/main/Maintainer.cpp

@@ -3,6 +3,8 @@
 // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
 
 #include "main/Maintainer.h"
+#include "herder/HerderPersistence.h"
+#include "ledger/LedgerHeaderUtils.h"
 #include "ledger/LedgerManager.h"
 #include "main/Application.h"
 #include "main/Config.h"
@@ -25,6 +27,7 @@ void
 Maintainer::start()
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
     auto& c = mApp.getConfig();
     if (c.AUTOMATIC_MAINTENANCE_PERIOD.count() > 0 &&
         c.AUTOMATIC_MAINTENANCE_COUNT > 0)
@@ -51,6 +54,7 @@ Maintainer::start()
 void
 Maintainer::scheduleMaintenance()
 {
+    releaseAssert(threadIsMain());
     mTimer.expires_from_now(mApp.getConfig().AUTOMATIC_MAINTENANCE_PERIOD);
     mTimer.async_wait([this]() { tick(); }, VirtualTimer::onFailureNoop);
 }
@@ -59,6 +63,7 @@ void
 Maintainer::tick()
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
     performMaintenance(mApp.getConfig().AUTOMATIC_MAINTENANCE_COUNT);
     scheduleMaintenance();
 }
@@ -67,6 +72,8 @@ void
 Maintainer::performMaintenance(uint32_t count)
 {
     ZoneScoped;
+    releaseAssert(threadIsMain());
+
     LOG_INFO(DEFAULT_LOG, "Performing maintenance");
     auto logSlow = LogSlowExecution(
         "Performing maintenance", LogSlowExecution::Mode::AUTOMATIC_RAII,
@@ -88,6 +95,25 @@ Maintainer::performMaintenance(uint32_t count)
 
     CLOG_INFO(History, "Trimming history <= ledger {}", lmin);
 
-    mApp.getLedgerManager().deleteOldEntries(mApp.getDatabase(), lmin, count);
+    // Cleanup SCP history, always from main
+    HerderPersistence::deleteOldEntries(mApp.getDatabase().getRawSession(),
+                                        lmin, count);
+
+    if (mApp.getConfig().parallelLedgerClose())
+    {
+        // Cleanup headers from background, to avoid conflicts with closing
+        // ledgers
+        mApp.postOnLedgerCloseThread(
+            [&db = mApp.getDatabase(), lmin, count]() {
+                auto session = std::make_unique<soci::session>(db.getPool());
+                LedgerHeaderUtils::deleteOldEntries(*session, lmin, count);
+            },
+            "maintenance: deleteOldEntries");
+    }
+    else
+    {
+        LedgerHeaderUtils::deleteOldEntries(mApp.getDatabase().getRawSession(),
+                                            lmin, count);
+    }
 }
 }