Adds HistoryArchiveUtils::readNextEntry<T> to read entries from an XDRInputFileStream with ledgerSeq bounds checking

Author: ThomasBrady

src/catchup/ApplyCheckpointWork.cpp

@@ -7,6 +7,7 @@
 #include "bucket/LiveBucketList.h"
 #include "catchup/ApplyLedgerWork.h"
 #include "history/FileTransferInfo.h"
+#include "history/HistoryArchiveUtils.h"
 #include "history/HistoryManager.h"
 #include "historywork/Progress.h"
 #include "ledger/CheckpointRange.h"
@@ -139,37 +140,21 @@ ApplyCheckpointWork::getCurrentTxSet()
     auto& lm = mApp.getLedgerManager();
     auto seq = lm.getLastClosedLedgerNum() + 1;
 
-    // Check mTxHistoryEntry prior to loading next history entry.
-    // This order is important because it accounts for ledger "gaps"
-    // in the history archives (which are caused by ledgers with empty tx
-    // sets, as those are not uploaded).
-    do
+    if (HistoryArchiveUtils::readNextEntry<TransactionHistoryEntry>(
+            mTxHistoryEntry, mTxIn, seq))
     {
-        if (mTxHistoryEntry.ledgerSeq < seq)
+        releaseAssert(mTxHistoryEntry.ledgerSeq == seq);
+        CLOG_DEBUG(History, "Loaded txset for ledger {}", seq);
+        if (mTxHistoryEntry.ext.v() == 0)
         {
-            CLOG_DEBUG(History, "Skipping txset for ledger {}",
-                       mTxHistoryEntry.ledgerSeq);
-        }
-        else if (mTxHistoryEntry.ledgerSeq > seq)
-        {
-            break;
+            return TxSetXDRFrame::makeFromWire(mTxHistoryEntry.txSet);
         }
         else
         {
-            releaseAssert(mTxHistoryEntry.ledgerSeq == seq);
-            CLOG_DEBUG(History, "Loaded txset for ledger {}", seq);
-            if (mTxHistoryEntry.ext.v() == 0)
-            {
-                return TxSetXDRFrame::makeFromWire(mTxHistoryEntry.txSet);
-            }
-            else
-            {
-                return TxSetXDRFrame::makeFromWire(
-                    mTxHistoryEntry.ext.generalizedTxSet());
-            }
+            return TxSetXDRFrame::makeFromWire(
+                mTxHistoryEntry.ext.generalizedTxSet());
         }
-    } while (mTxIn && mTxIn.readOne(mTxHistoryEntry));
-
+    }
     CLOG_DEBUG(History, "Using empty txset for ledger {}", seq);
     return TxSetXDRFrame::makeEmpty(lm.getLastClosedLedgerHeader());
 }
@@ -181,30 +166,12 @@ ApplyCheckpointWork::getCurrentTxResultSet()
     ZoneScoped;
     auto& lm = mApp.getLedgerManager();
     auto seq = lm.getLastClosedLedgerNum() + 1;
-    // Check mTxResultSet prior to loading next result set.
-    // This order is important because it accounts for ledger "gaps"
-    // in the history archives (which are caused by ledgers with empty tx
-    // sets, as those are not uploaded).
-    while (mTxResultIn && mTxResultIn->readOne(*mTxHistoryResultEntry))
+    if (HistoryArchiveUtils::readNextEntry<TransactionHistoryResultEntry>(
+            *mTxHistoryResultEntry, *mTxResultIn, seq))
     {
-        if (mTxHistoryResultEntry)
-        {
-            if (mTxHistoryResultEntry->ledgerSeq < seq)
-            {
-                CLOG_DEBUG(History, "Advancing past txresultset for ledger {}",
-                           mTxHistoryResultEntry->ledgerSeq);
-            }
-            else if (mTxHistoryResultEntry->ledgerSeq > seq)
-            {
-                break;
-            }
-            else
-            {
-                releaseAssert(mTxHistoryResultEntry->ledgerSeq == seq);
-                CLOG_DEBUG(History, "Loaded txresultset for ledger {}", seq);
-                return std::make_optional(mTxHistoryResultEntry->txResultSet);
-            }
-        }
+        releaseAssert(mTxHistoryResultEntry->ledgerSeq == seq);
+        CLOG_DEBUG(History, "Loaded txresultset for ledger {}", seq);
+        return std::make_optional(mTxHistoryResultEntry->txResultSet);
     }
     CLOG_DEBUG(History, "No txresultset for ledger {}", seq);
     return std::nullopt;

src/history/HistoryArchiveUtils.h

@@ -0,0 +1,60 @@
+#pragma once
+
+#include "util/XDRStream.h"
+#include <functional>
+#include <type_traits>
+
+namespace stellar
+{
+
+// Type trait to check for ledgerSeq member.
+template<typename T, typename = void>
+struct has_ledger_seq : std::false_type {};
+template<typename T>
+struct has_ledger_seq<T, std::void_t<decltype(std::declval<T>().ledgerSeq)>> 
+    : std::true_type {};
+
+class HistoryArchiveUtils
+{
+public:
+    template<typename T>
+    using ValidationFunc = std::function<bool(T const&)>;
+
+    // Read next entry from XDRInputFileStream at targetLedger.
+    // Returns true if an entry matching targetLedger was found,
+    // false otherwise. 
+    // Callers must ensure that the stream is open, and reuse the same
+    // entry object for each call.
+    // ValidationFunc<T>, if provided, is called on each entry
+    // to validate it immediately after reading it from the stream.
+    template<typename T, typename = std::enable_if_t<has_ledger_seq<T>::value>>
+    static bool readNextEntry(T& entry, XDRInputFileStream& in, uint32_t targetLedger,
+                            std::optional<ValidationFunc<T>> validationFunc = std::nullopt)
+    {
+
+        do 
+        {
+            if (entry.ledgerSeq < targetLedger)
+            {
+                CLOG_DEBUG(History, "Advancing past ledger {} before target ledger {}",
+                            entry.ledgerSeq, targetLedger);
+                // Skip entries before targetLedger.
+                continue;
+            }
+            else if (entry.ledgerSeq > targetLedger)
+            {
+                // No entry for this ledger.
+                break;
+            }
+            else
+            {
+                // Found matching entry.
+                return true;
+            }
+        } while (in && in.readOne(entry) && (validationFunc ? (*validationFunc)(entry) : true));
+
+        return false;
+    }
+};
+
+} // namespace stellar

src/historywork/VerifyTxResultsWork.cpp

@@ -4,6 +4,7 @@
 
 #include "historywork/VerifyTxResultsWork.h"
 #include "history/FileTransferInfo.h"
+#include "history/HistoryArchiveUtils.h"
 #include "ledger/LedgerManager.h"
 #include "main/ErrorMessages.h"
 #include "util/FileSystemException.h"
@@ -141,51 +142,33 @@ TransactionHistoryResultEntry
 VerifyTxResultsWork::getCurrentTxResultSet(uint32_t ledger)
 {
     ZoneScoped;
+
     TransactionHistoryResultEntry trs;
     trs.ledgerSeq = ledger;
+    auto validateEntry = [this](TransactionHistoryResultEntry const& entry) {
+        auto readLedger = entry.ledgerSeq;
+        auto low = HistoryManager::firstLedgerInCheckpointContaining(
+            mCheckpoint, mApp.getConfig());
 
-    auto readNextWithValidation = [&]() {
-        auto res = mResIn.readOne(mTxResultEntry);
-        if (res)
-        {
-            auto readLedger = mTxResultEntry.ledgerSeq;
-            auto low = HistoryManager::firstLedgerInCheckpointContaining(
-                mCheckpoint, mApp.getConfig());
-            if (readLedger > mCheckpoint || readLedger < low)
-            {
-                throw std::runtime_error("Results outside of checkpoint range");
-            }
-
-            if (readLedger <= mLastSeenLedger)
-            {
-                throw std::runtime_error("Malformed or duplicate results: "
-                                         "ledgers must be strictly increasing");
-            }
-            mLastSeenLedger = readLedger;
-        }
-        return res;
-    };
-
-    do
-    {
-        if (mTxResultEntry.ledgerSeq < ledger)
-        {
-            CLOG_DEBUG(History, "Processed tx results for ledger {}",
-                       mTxResultEntry.ledgerSeq);
-        }
-        else if (mTxResultEntry.ledgerSeq > ledger)
+        if (readLedger > mCheckpoint || readLedger < low)
         {
-            // No tx results in this ledger
-            break;
+            throw std::runtime_error("Results outside of checkpoint range");
         }
-        else
+        if (readLedger <= mLastSeenLedger)
         {
-            CLOG_DEBUG(History, "Loaded tx result set for ledger {}", ledger);
-            trs.txResultSet = mTxResultEntry.txResultSet;
-            return trs;
+            throw std::runtime_error("Malformed or duplicate results: "
+                                     "ledgers must be strictly increasing");
         }
-    } while (mResIn && readNextWithValidation());
+        mLastSeenLedger = readLedger;
+        return true;
+    };
 
+    if (HistoryArchiveUtils::readNextEntry<TransactionHistoryResultEntry>(
+            mTxResultEntry, mResIn, ledger, validateEntry))
+    {
+        CLOG_DEBUG(History, "Loaded tx result set for ledger {}", ledger);
+        trs.txResultSet = mTxResultEntry.txResultSet;
+    }
     return trs;
 }
 }