Fixes after rebase

Author: SirTyson

src/bucket/BucketUtils.cpp

@@ -147,10 +147,10 @@ EvictionResultCandidates::isValid(uint32_t currLedgerSeq,
     // affect evictions scans.
     if (protocolVersionIsBefore(
             initialLedgerVers,
-            BucketBase::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION) &&
+            LiveBucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION) &&
         protocolVersionStartsFrom(
             currLedgerVers,
-            BucketBase::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION))
+            LiveBucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION))
     {
         return false;
     }

src/history/test/HistoryTests.cpp

@@ -1250,10 +1250,10 @@ TEST_CASE("Catchup with protocol upgrade", "[catchup][history]")
     {
         if (protocolVersionEquals(
                 Config::CURRENT_LEDGER_PROTOCOL_VERSION,
-                BucketBase::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION))
+                LiveBucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION))
         {
             testUpgrade(
-                BucketBase::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION);
+                LiveBucket::FIRST_PROTOCOL_SUPPORTING_PERSISTENT_EVICTION);
         }
     }
 }

src/historywork/DownloadBucketsWork.cpp

@@ -11,7 +11,6 @@
 #include "historywork/GetAndUnzipRemoteFileWork.h"
 #include "historywork/VerifyBucketWork.h"
 #include "work/WorkWithCallback.h"
-#include "xdr/Stellar-contract-config-setting.h"
 #include <Tracy.hpp>
 #include <fmt/format.h>
 
@@ -73,13 +72,13 @@ DownloadBucketsWork::resetIter()
     mNextHotBucketIter = mHotHashes.begin();
 }
 
-template<typename BucketT>
+template <typename BucketT>
 bool
 DownloadBucketsWork::onSuccessCb(
     Application& app, FileTransferInfo const& ft, std::string const& hash,
-    int currId,
-    std::map<std::string, std::shared_ptr<BucketT>>& buckets,
-    std::map<int, std::unique_ptr<typename BucketT::IndexT const>>& indexMap)
+    int currId, std::map<std::string, std::shared_ptr<BucketT>>& buckets,
+    std::map<int, std::unique_ptr<typename BucketT::IndexT const>>& indexMap,
+    std::lock_guard<std::mutex> const& indexMapLock)
 {
     auto bucketPath = ft.localPath_nogz();
     auto indexIter = indexMap.find(currId);
@@ -140,29 +139,47 @@ DownloadBucketsWork::yieldMoreWork()
     if (isHotHash)
     {
         auto currId = mHotIndexId++;
+        mHotIndexMapMutex.lock();
         auto [indexIter, inserted] = mHotIndexMap.emplace(currId, nullptr);
+        mHotIndexMapMutex.unlock();
         releaseAssertOrThrow(inserted);
         verifyWork = std::make_shared<VerifyBucketWork<HotArchiveBucket>>(mApp, ft.localPath_nogz(),
                                                 hexToBin256(hash),
                                                 indexIter->second, failureCb);
-        adoptBucketCb = [this, &ft, hash, currId](Application& app) {
-            return onSuccessCb<HotArchiveBucket>(app, ft, hash, currId,
-                                               mHotBuckets, mHotIndexMap);
+        adoptBucketCb = [weakSelf, ft, hash, currId](Application& app) {
+            auto self = weakSelf.lock();
+            if (self)
+            {
+                std::lock_guard lock(self->mHotIndexMapMutex);
+                return onSuccessCb<HotArchiveBucket>(app, ft, hash, currId,
+                                                     self->mHotBuckets,
+                                                     self->mHotIndexMap, lock);
+            }
+            return true;
         };
 
         mNextHotBucketIter++;
     }
     else
     {
         auto currId = mLiveIndexId++;
+        mLiveIndexMapMutex.lock();
         auto [indexIter, inserted] = mLiveIndexMap.emplace(currId, nullptr);
+        mLiveIndexMapMutex.unlock();
         releaseAssertOrThrow(inserted);
         verifyWork = std::make_shared<VerifyBucketWork<LiveBucket>>(mApp, ft.localPath_nogz(),
                                                 hexToBin256(hash),
                                                 indexIter->second, failureCb);
-        adoptBucketCb = [this, &ft, hash, currId](Application& app) {
-            return onSuccessCb<LiveBucket>(app, ft, hash, currId,
-                                               mLiveBuckets, mLiveIndexMap);
+        adoptBucketCb = [weakSelf, ft, hash, currId](Application& app) {
+            auto self = weakSelf.lock();
+            if (self)
+            {
+                std::lock_guard lock(self->mLiveIndexMapMutex);
+                return onSuccessCb<LiveBucket>(app, ft, hash, currId,
+                                               self->mLiveBuckets,
+                                               self->mLiveIndexMap, lock);
+            }
+            return true;
         };
 
         mNextLiveBucketIter++;
@@ -181,10 +198,12 @@ DownloadBucketsWork::yieldMoreWork()
 template bool DownloadBucketsWork::onSuccessCb<LiveBucket>(
     Application&, FileTransferInfo const&, std::string const&, int,
     std::map<std::string, std::shared_ptr<LiveBucket>>&,
-    std::map<int, std::unique_ptr<LiveBucketIndex const>>&);
+    std::map<int, std::unique_ptr<LiveBucketIndex const>>&,
+    std::lock_guard<std::mutex> const&);
 
 template bool DownloadBucketsWork::onSuccessCb<HotArchiveBucket>(
     Application&, FileTransferInfo const&, std::string const&, int,
     std::map<std::string, std::shared_ptr<HotArchiveBucket>>&,
-    std::map<int, std::unique_ptr<HotArchiveBucketIndex const>>&);
+    std::map<int, std::unique_ptr<HotArchiveBucketIndex const>>&,
+    std::lock_guard<std::mutex> const&);
 }

src/historywork/DownloadBucketsWork.h

@@ -27,9 +27,14 @@ class DownloadBucketsWork : public BatchWork
     TmpDir const& mDownloadDir;
     std::shared_ptr<HistoryArchive> mArchive;
 
-    // Store indexes of downloaded buckets
+    // Store indexes of downloaded buckets. Child processes will actually create
+    // the indexes, but DownloadBucketsWork needs to maintain actual ownership
+    // of the pointers so that the success callback can pass them to the
+    // BucketManager. Must be protected by a mutex to avoid race conditions.
     std::map<int, std::unique_ptr<LiveBucketIndex const>> mLiveIndexMap;
     std::map<int, std::unique_ptr<HotArchiveBucketIndex const>> mHotIndexMap;
+    std::mutex mLiveIndexMapMutex;
+    std::mutex mHotIndexMapMutex;
     int mLiveIndexId{0};
     int mHotIndexId{0};
 
@@ -39,7 +44,8 @@ class DownloadBucketsWork : public BatchWork
                 std::string const& hash, int currId,
                 std::map<std::string, std::shared_ptr<BucketT>>& buckets,
                 std::map<int, std::unique_ptr<typename BucketT::IndexT const>>&
-                    indexMap);
+                    indexMap,
+                std::lock_guard<std::mutex> const& indexMapLock);
 
   public:
     // Note: hashes must contain both live and hot archive bucket hashes