WIP - spike on hot archive adaptation

Author: sisuresh

src/ledger/LedgerManagerImpl.cpp

@@ -1617,7 +1617,7 @@ LedgerManagerImpl::collectEntries(AppConnector& app, AbstractLedgerTxn& ltx,
     return entryMap;
 }
 
-void
+RestoredKeys
 LedgerManagerImpl::applyThread(AppConnector& app, ThreadEntryMap& entryMap,
                                Thread const& thread, Config const& config,
                                SorobanNetworkConfig const& sorobanConfig,
@@ -1629,6 +1629,7 @@ LedgerManagerImpl::applyThread(AppConnector& app, ThreadEntryMap& entryMap,
     // we accumulate the RO extensions until we need to apply them.
     UnorderedMap<LedgerKey, uint32_t> readOnlyTtlExtensions;
 
+    RestoredKeys threadRestoredKeys;
     for (auto const& txBundle : thread)
     {
         releaseAssertOrThrow(txBundle.getResPayload());
@@ -1676,7 +1677,7 @@ LedgerManagerImpl::applyThread(AppConnector& app, ThreadEntryMap& entryMap,
             txBundle.getResPayload(), sorobanMetrics, txSubSeed,
             txBundle.getEffects());
 
-        if (res.mSuccess)
+        if (res.getSuccess())
         {
             UnorderedMap<LedgerKey, bool> isReadOnlyTTLMap;
             for (auto const& ro :
@@ -1699,7 +1700,7 @@ LedgerManagerImpl::applyThread(AppConnector& app, ThreadEntryMap& entryMap,
             }
 
             // now apply the entry changes to entryMap
-            for (auto const& entry : res.mModifiedEntryMap)
+            for (auto const& entry : res.getModifiedEntryMap())
             {
                 auto const& lk = entry.first;
                 auto const& updatedLe = entry.second;
@@ -1738,11 +1739,24 @@ LedgerManagerImpl::applyThread(AppConnector& app, ThreadEntryMap& entryMap,
                     it->second = {updatedLe, true};
                 }
             }
+
+            for (auto const& key : res.getRestoredKeys().hotArchive)
+            {
+                auto [_, inserted] = threadRestoredKeys.hotArchive.emplace(key);
+                releaseAssertOrThrow(inserted);
+            }
+            for (auto const& key : res.getRestoredKeys().liveBucketList)
+            {
+                auto [_, inserted] =
+                    threadRestoredKeys.liveBucketList.emplace(key);
+                releaseAssertOrThrow(inserted);
+            }
         }
         else
         {
             releaseAssertOrThrow(!txBundle.getResPayload()->isSuccess());
         }
+        return threadRestoredKeys;
     }
 
     for (auto kvp : readOnlyTtlExtensions)
@@ -1798,6 +1812,8 @@ LedgerManagerImpl::applySorobanStage(AppConnector& app, AbstractLedgerTxn& ltx,
         entryMapsByThread.emplace_back(collectEntries(app, ltx, thread));
     }
 
+    std::vector<std::future<RestoredKeys>> threadRestoredKeyFutures;
+
     auto ledgerInfo = getParallelLedgerInfo(app, ltx);
     std::vector<std::thread> threads;
     for (size_t i = 0; i < stage.size(); ++i)
@@ -1806,19 +1822,51 @@ LedgerManagerImpl::applySorobanStage(AppConnector& app, AbstractLedgerTxn& ltx,
 
         auto const& thread = stage.at(i);
 
-        threads.emplace_back(&LedgerManagerImpl::applyThread, this,
+        /* threads.emplace_back(&LedgerManagerImpl::applyThread, this,
                              std::ref(app), std::ref(entryMapByThread),
                              std::ref(thread), config, sorobanConfig,
                              std::ref(ledgerInfo), sorobanBasePrngSeed,
                              std::ref(sorobanMetrics));
+ */
+        threadRestoredKeyFutures.emplace_back(std::async(
+            std::launch::async, &LedgerManagerImpl::applyThread, this,
+            std::ref(app), std::ref(entryMapByThread), std::ref(thread), config,
+            sorobanConfig, std::ref(ledgerInfo), sorobanBasePrngSeed,
+            std::ref(sorobanMetrics)));
     }
 
     for (auto& thread : threads)
     {
         thread.join();
     }
 
+    std::vector<RestoredKeys> threadRestoredKeys;
+    for (auto& restoredKeysFutures : threadRestoredKeyFutures)
+    {
+        threadRestoredKeys.emplace_back(restoredKeysFutures.get());
+    }
+
     LedgerTxn ltxInner(ltx);
+    for (auto const& restoredKeys : threadRestoredKeys)
+    {
+        // We filter out the TTL keys here because LedgerTxn will add the ttl
+        // key
+        for (auto const& key : restoredKeys.hotArchive)
+        {
+            if (key.type() != TTL)
+            {
+                ltxInner.addRestoredFromHotArchiveKey(key);
+            }
+        }
+        for (auto const& key : restoredKeys.liveBucketList)
+        {
+            if (key.type() != TTL)
+            {
+                ltxInner.addRestoredFromLiveBucketListKey(key);
+            }
+        }
+    }
+
     for (auto const& thread : stage)
     {
         for (auto const& txBundle : thread)
@@ -1886,6 +1934,11 @@ LedgerManagerImpl::applySorobanStage(AppConnector& app, AbstractLedgerTxn& ltx,
                         auto newLiveUntil =
                             updatedEntry.data.ttl().liveUntilLedgerSeq;
 
+                        // The only scenario where we accept a reduction in TTL
+                        // is one where an entry was deleted, and then
+                        // recreated. This can only happen if the key was in the
+                        // readWrite set, so if it's not then this is just a
+                        // parallel readOnly bump that we can ignore here.
                         if (newLiveUntil <= currLiveUntil &&
                             isInReadWriteSet.count(entry.first) == 0)
                         {

src/ledger/LedgerManagerImpl.h

@@ -122,12 +122,13 @@ class LedgerManagerImpl : public LedgerManager
     ThreadEntryMap collectEntries(AppConnector& app, AbstractLedgerTxn& ltx,
                                   Thread const& txs);
 
-    void applyThread(AppConnector& app, ThreadEntryMap& entryMapByCluster,
-                     Thread const& thread, Config const& config,
-                     SorobanNetworkConfig const& sorobanConfig,
-                     ParallelLedgerInfo const& ledgerInfo,
-                     Hash const& sorobanBasePrngSeed,
-                     SorobanMetrics& sorobanMetrics);
+    RestoredKeys applyThread(AppConnector& app,
+                             ThreadEntryMap& entryMapByCluster,
+                             Thread const& thread, Config const& config,
+                             SorobanNetworkConfig const& sorobanConfig,
+                             ParallelLedgerInfo const& ledgerInfo,
+                             Hash const& sorobanBasePrngSeed,
+                             SorobanMetrics& sorobanMetrics);
 
     void applySorobanStage(AppConnector& app, AbstractLedgerTxn& ltx,
                            ApplyStage const& stage,

src/ledger/LedgerTxn.cpp

@@ -825,6 +825,68 @@ LedgerTxn::Impl::erase(InternalLedgerKey const& key)
     }
 }
 
+void
+LedgerTxn::addRestoredFromHotArchiveKey(LedgerKey const& key)
+{
+    getImpl()->addRestoredFromHotArchiveKey(key);
+}
+
+void
+LedgerTxn::Impl::addRestoredFromHotArchiveKey(LedgerKey const& key)
+{
+    throwIfSealed();
+    throwIfChild();
+
+    if (!isPersistentEntry(key))
+    {
+        throw std::runtime_error("Key type not supported in Hot Archive");
+    }
+
+    auto ttlKey = getTTLKey(key);
+
+    // Mark the keys as restored
+    auto addKey = [this](LedgerKey const& key) {
+        auto [_, inserted] = mRestoredKeys.hotArchive.insert(key);
+        if (!inserted)
+        {
+            throw std::runtime_error("Key already removed from hot archive");
+        }
+    };
+    addKey(key);
+    addKey(ttlKey);
+}
+
+void
+LedgerTxn::addRestoredFromLiveBucketListKey(LedgerKey const& key)
+{
+    getImpl()->addRestoredFromLiveBucketListKey(key);
+}
+
+void
+LedgerTxn::Impl::addRestoredFromLiveBucketListKey(LedgerKey const& key)
+{
+    throwIfSealed();
+    throwIfChild();
+
+    if (!isPersistentEntry(key))
+    {
+        throw std::runtime_error("Key type not supported for restoration");
+    }
+
+    auto ttlKey = getTTLKey(key);
+
+    auto addKey = [this](LedgerKey const& key) {
+        auto [_, inserted] = mRestoredKeys.liveBucketList.insert(key);
+        if (!inserted)
+        {
+            throw std::runtime_error(
+                "Key already restored from Live BucketList");
+        }
+    };
+    addKey(key);
+    addKey(ttlKey);
+}
+
 void
 LedgerTxn::restoreFromHotArchive(LedgerEntry const& entry, uint32_t ttl)
 {

src/ledger/LedgerTxn.h

@@ -595,6 +595,8 @@ class AbstractLedgerTxn : public AbstractLedgerTxnParent
                                        uint32_t ttl) = 0;
     virtual void restoreFromLiveBucketList(LedgerKey const& key,
                                            uint32_t ttl) = 0;
+    virtual void addRestoredFromHotArchiveKey(LedgerKey const& key) = 0;
+    virtual void addRestoredFromLiveBucketListKey(LedgerKey const& key) = 0;
     virtual LedgerTxnEntry load(InternalLedgerKey const& key) = 0;
     virtual ConstLedgerTxnEntry
     loadWithoutRecord(InternalLedgerKey const& key) = 0;
@@ -742,6 +744,8 @@ class LedgerTxn : public AbstractLedgerTxn
     void erase(InternalLedgerKey const& key) override;
     void restoreFromHotArchive(LedgerEntry const& entry, uint32_t ttl) override;
     void restoreFromLiveBucketList(LedgerKey const& key, uint32_t ttl) override;
+    void addRestoredFromHotArchiveKey(LedgerKey const& key) override;
+    void addRestoredFromLiveBucketListKey(LedgerKey const& key) override;
 
     UnorderedMap<LedgerKey, LedgerEntry> getAllOffers() override;
 

src/ledger/LedgerTxnImpl.h

@@ -361,6 +361,18 @@ class LedgerTxn::Impl
     // - the entry cache may be, but is not guaranteed to be, cleared.
     void erase(InternalLedgerKey const& key);
 
+    // addRestoredFromHotArchiveKey has the basic exception safety guarantee. If
+    // it throws an exception, then
+    // - the prepared statement cache may be, but is not guaranteed to be,
+    //   modified
+    void addRestoredFromHotArchiveKey(LedgerKey const& key);
+
+    // addRestoredFromLiveBucketListKey has the basic exception safety
+    // guarantee. If it throws an exception, then
+    // - the prepared statement cache may be, but is not guaranteed to be,
+    //   modified
+    void addRestoredFromLiveBucketListKey(LedgerKey const& key);
+
     // restoreFromHotArchive has the basic exception safety guarantee. If it
     // throws an exception, then
     // - the prepared statement cache may be, but is not guaranteed to be,

src/ledger/test/LedgerCloseMetaStreamTests.cpp

@@ -421,8 +421,8 @@ TEST_CASE_VERSIONS("meta stream contains reasonable meta", "[ledgerclosemeta]")
                             .bucketListWindowSamplePeriod = 1;
                     });
 
-                // Modify Soroban network config closes a ledger
-                ++targetSeq;
+                // Modify Soroban network config closes 2 ledgers
+                targetSeq += 2;
             }
 
             auto root = TestAccount::createRoot(*app);