Resume ballot on top of overlay v2

src/herder/HerderSCPDriver.cpp

@@ -1319,6 +1319,14 @@ HerderSCPDriver::recordBallotBlockedOnTxSet(uint64_t slotIndex,
         timing.mBallotBlockedOnTxSetStart.end())
     {
         timing.mBallotBlockedOnTxSetStart[value] = mApp.getClock().now();
+
+        if (StellarValue sv;
+            isParallelTxSetDownloadEnabled() && toStellarValue(value, sv))
+        {
+            // Remember that `value` is stalled waiting for the tx set
+            // with hash `sv.txSetHash`.
+            mStallingByTxSet[sv.txSetHash].emplace_back(slotIndex, value);
+        }
     }
 }
 
@@ -1337,6 +1345,27 @@ HerderSCPDriver::measureAndRecordBallotBlockedOnTxSet(uint64_t slotIndex,
                 std::chrono::duration_cast<std::chrono::milliseconds>(
                     mApp.getClock().now() - valueIt->second);
             mSCPMetrics.mBallotBlockedOnTxSet.Update(elapsed);
+
+            if (StellarValue sv; toStellarValue(value, sv))
+            {
+                // This value is no longer stalled. Remove it from
+                // `mStallingByTxSet`
+                auto sIt = mStallingByTxSet.find(sv.txSetHash);
+                if (sIt != mStallingByTxSet.end())
+                {
+                    auto& vec = sIt->second;
+                    vec.erase(std::remove_if(vec.begin(), vec.end(),
+                                             [&](auto const& p) {
+                                                 return p.first == slotIndex &&
+                                                        p.second == value;
+                                             }),
+                              vec.end());
+                    if (vec.empty())
+                    {
+                        mStallingByTxSet.erase(sIt);
+                    }
+                }
+            }
             return;
         }
     }
@@ -1695,6 +1724,23 @@ HerderSCPDriver::purgeSlotsOutsideRange(std::optional<uint64_t> minSlotIndex,
     // Clean up expired weak_ptrs from the pending tx set registries.
     purgeExpiredWeakPtrs(mPendingTxSetWrappers);
     purgeExpiredWeakPtrs(mPendingTxSetEnvelopeWrappers);
+
+    // Drop stalled-ballot resume entries whose slots fall outside the retained
+    // range.
+    for (auto it = mStallingByTxSet.begin(); it != mStallingByTxSet.end();)
+    {
+        auto& stalling = it->second;
+        stalling.erase(
+            std::remove_if(stalling.begin(), stalling.end(),
+                           [&](auto const& slotAndValue) {
+                               auto const slot = slotAndValue.first;
+                               return slot != slotToKeep &&
+                                      ((minSlotIndex && slot < *minSlotIndex) ||
+                                       (maxSlotIndex && slot > *maxSlotIndex));
+                           }),
+            stalling.end());
+        it = stalling.empty() ? mStallingByTxSet.erase(it) : std::next(it);
+    }
 }
 
 void
@@ -1728,6 +1774,34 @@ HerderSCPDriver::onTxSetReceived(Hash const& txSetHash,
         }
         mPendingTxSetEnvelopeWrappers.erase(envIt);
     }
+
+    // Resume any slot that stalled waiting for this tx set
+    maybeResumeBalloting(txSetHash);
+}
+
+void
+HerderSCPDriver::maybeResumeBalloting(Hash const& txSetHash)
+{
+    if (!isParallelTxSetDownloadEnabled())
+    {
+        return;
+    }
+
+    auto it = mStallingByTxSet.find(txSetHash);
+    if (it == mStallingByTxSet.end())
+    {
+        return;
+    }
+
+    // Remove entry from `mStallingByTxSet` and iterate over stalling slots
+    // Remove prior to iterating because the `receivedTxSet` flow may itself
+    // modify `mStallingByTxSet`.
+    auto const stalling = std::move(it->second);
+    mStallingByTxSet.erase(it);
+    for (auto const& slotAndValue : stalling)
+    {
+        mSCP.receivedTxSet(slotAndValue.first, slotAndValue.second);
+    }
 }
 
 void

src/herder/HerderSCPDriver.h

@@ -175,6 +175,10 @@ class HerderSCPDriver : public SCPDriver
     // downloading).
     void onTxSetReceived(Hash const& txSetHash, TxSetXDRFrameConstPtr txSet);
 
+    // If balloting is stalled waiting for txSetHash, then resume balloting from
+    // the stall point. Otherwise, do nothing.
+    void maybeResumeBalloting(Hash const& txSetHash);
+
     double getExternalizeLag(NodeID const& id) const;
 
     Json::Value getQsetLagInfo(bool summary, bool fullKeys);
@@ -313,6 +317,11 @@ class HerderSCPDriver : public SCPDriver
     // * first prepare to externalize
     std::map<uint64_t, SCPTiming> mSCPExecutionTimes;
 
+    // Values stalled at the ballot commit gate waiting for a tx set.
+    // Mapping from <tx set hashes> to pairs of (<slot index>, <stalled value>).
+    UnorderedMap<Hash, std::vector<std::pair<uint64_t, Value>>>
+        mStallingByTxSet;
+
     uint32_t mLedgerSeqNominating;
     ValueWrapperPtr mCurrentValue;
 

src/scp/BallotProtocol.cpp

@@ -54,7 +54,7 @@ BallotProtocol::isNewerStatement(NodeID const& nodeID, SCPStatement const& st)
 
 bool
 BallotProtocol::isNewerStatement(SCPStatement const& oldst,
-                                 SCPStatement const& st)
+                                 SCPStatement const& st) const
 {
     bool res = false;
 
@@ -98,7 +98,7 @@ BallotProtocol::isNewerStatement(SCPStatement const& oldst,
         else
         {
             // Lexicographical order between PREPARE statements:
-            // (b, p, p', h)
+            // (b, p, p', h, c)
             auto const& oldPrep = oldst.pledges.prepare();
             auto const& prep = st.pledges.prepare();
 
@@ -124,7 +124,16 @@ BallotProtocol::isNewerStatement(SCPStatement const& oldst,
                     }
                     else if (compBallot == 0)
                     {
-                        res = (oldPrep.nH < prep.nH);
+                        if (mSlot.getSCPDriver()
+                                .protocolAllowsEmptyTxSetValues() &&
+                            oldPrep.nH == prep.nH)
+                        {
+                            res = (oldPrep.nC < prep.nC);
+                        }
+                        else
+                        {
+                            res = (oldPrep.nH < prep.nH);
+                        }
                     }
                 }
             }
@@ -672,7 +681,7 @@ BallotProtocol::createStatement(SCPStatementType const& type)
     return statement;
 }
 
-void
+SCPStatement
 BallotProtocol::emitCurrentStateStatement()
 {
     ZoneScoped;
@@ -726,6 +735,11 @@ BallotProtocol::emitCurrentStateStatement()
             throw std::runtime_error("moved to a bad state (ballot protocol)");
         }
     }
+
+    // Return the statement this call generated. Intentionally does not return
+    // the statement generated by recursion so that the caller may reason about
+    // what this call specifically produced.
+    return envelope.statement;
 }
 
 void
@@ -1141,6 +1155,7 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH)
                mSlot.getSlotIndex(), mSlot.getSCP().ballotToStr(newH));
 
     bool didWork = false;
+    bool stalled = false;
 
     // remember newH's value
     mValueOverride = mSlot.getSCPDriver().wrapValue(newH.value);
@@ -1176,6 +1191,8 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH)
                 mSlot.getSCPDriver().recordBallotBlockedOnTxSet(
                     mSlot.getSlotIndex(), newC.value);
 
+                stalled = true;
+
                 CLOG_TRACE(
                     SCP,
                     "BallotProtocol::setConfirmPrepared slot:{} "
@@ -1214,12 +1231,53 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH)
 
     if (didWork)
     {
-        emitCurrentStateStatement();
+        auto const emitted = emitCurrentStateStatement();
+
+        if (stalled)
+        {
+            // Stalled waiting for the tx set corresponding to newC.value.
+            // Remember the state that existed at the stall point so that we can
+            // evaluate whether it's safe to resume (skipping a ballot timeout)
+            // if the tx set arrives.
+            mStalledCommit = StalledCommit{newC, newH, emitted};
+        }
+        else
+        {
+            mStalledCommit.reset();
+        }
     }
 
     return didWork;
 }
 
+void
+BallotProtocol::receivedTxSet(Value const& value)
+{
+    ZoneScoped;
+    // Only act if this slot stalled waiting for exactly this value's tx set.
+    if (!mStalledCommit || !(mStalledCommit->mCommitBallot.value == value))
+    {
+        return;
+    }
+
+    // It should not be possible to end up here prior to the protocol supporting
+    // kStructurallyValidValue.
+    releaseAssert(mSlot.getSCPDriver().protocolAllowsEmptyTxSetValues());
+
+    auto const stalled = *mStalledCommit;
+    mStalledCommit.reset();
+
+    // Resume only if the node has done no balloting work since the stall.
+    auto const* selfEnv = getLatestMessage(mSlot.getSCP().getLocalNodeID());
+    if (selfEnv == nullptr || !(selfEnv->statement == stalled.mStallStatement))
+    {
+        return;
+    }
+
+    // Re-run the commit step setConfirmPrepared deferred.
+    setConfirmPrepared(stalled.mCommitBallot, stalled.mHighBallot);
+}
+
 void
 BallotProtocol::findExtendedInterval(Interval& candidate,
                                      std::set<uint32> const& boundaries,

src/scp/BallotProtocol.h

@@ -9,6 +9,7 @@
 #include "util/GlobalChecks.h"
 #include <functional>
 #include <memory>
+#include <optional>
 #include <set>
 #include <string>
 #include <utility>
@@ -95,6 +96,16 @@ class BallotProtocol
     SCPEnvelopeWrapperPtr
         mLastEnvelopeEmit; // last envelope emitted by this node
 
+    // Information about the state of balloting upon stalling when attempting to
+    // set `c` to a value the node has not successfully fetched.
+    struct StalledCommit
+    {
+        SCPBallot mCommitBallot;      // c (deferred; not in the emitted stmt)
+        SCPBallot mHighBallot;        // h
+        SCPStatement mStallStatement; // self statement emitted at the stall
+    };
+    std::optional<StalledCommit> mStalledCommit;
+
   public:
     BallotProtocol(Slot& slot);
 
@@ -119,6 +130,12 @@ class BallotProtocol
     // flavor that takes the actual desired counter value
     bool bumpState(Value const& value, uint32 n);
 
+    // Called when the tx set referenced by @p value arrives.
+    // If balloting stalled waiting for this tx set, AND the node's state has
+    // not changed since hitting the stall point, this function will resume
+    // balloting for this slot immediately. Otherwise, it does nothing.
+    void receivedTxSet(Value const& value);
+
     // ** status methods
 
     // returns information about the local state in JSON format
@@ -166,8 +183,8 @@ class BallotProtocol
     static std::set<Value> getStatementValues(SCPStatement const& st);
 
     // returns true if st is newer than oldst
-    static bool isNewerStatement(SCPStatement const& oldst,
-                                 SCPStatement const& st);
+    bool isNewerStatement(SCPStatement const& oldst,
+                          SCPStatement const& st) const;
 
   private:
     // attempts to make progress using the latest statement as a hint
@@ -305,9 +322,11 @@ class BallotProtocol
     // we have.
     bool updateCurrentValue(SCPBallot const& ballot);
 
-    // emits a statement reflecting the nodes' current state
-    // and attempts to make progress
-    void emitCurrentStateStatement();
+    // Emits a statement reflecting the node's current state and attempts to
+    // make progress. Returns the statement it generated (built from current
+    // state, *before* the self-processing recursion that may advance the ballot
+    // further), so callers can capture exactly what this call produced.
+    SCPStatement emitCurrentStateStatement();
 
     // verifies that the internal state is consistent
     void checkInvariants();

src/scp/SCP.cpp

@@ -51,6 +51,16 @@ SCP::stopNomination(uint64 slotIndex)
     }
 }
 
+void
+SCP::receivedTxSet(uint64 slotIndex, Value const& value)
+{
+    auto s = getSlot(slotIndex, false);
+    if (s)
+    {
+        s->receivedTxSet(value);
+    }
+}
+
 void
 SCP::updateLocalQuorumSet(SCPQuorumSet const& qSet)
 {

src/scp/SCP.h

@@ -59,6 +59,10 @@ class SCP
     // stops nomination for a slot
     void stopNomination(uint64 slotIndex);
 
+    // Notify SCP that the tx set referenced by @p value has arrived so that it
+    // may resume balloting if stalled waiting for this tx set.
+    void receivedTxSet(uint64 slotIndex, Value const& value);
+
     // Local QuorumSet interface (can be dynamically updated)
     void updateLocalQuorumSet(SCPQuorumSet const& qSet);
     SCPQuorumSet const& getLocalQuorumSet();

src/scp/Slot.cpp

@@ -131,7 +131,7 @@ Slot::isNewerNominationOrBallotSt(SCPStatement const& oldSt,
     }
     else
     {
-        if (BallotProtocol::isNewerStatement(oldSt, newSt))
+        if (mBallotProtocol.isNewerStatement(oldSt, newSt))
         {
             replace = true;
         }
@@ -217,6 +217,12 @@ Slot::abandonBallot()
     return mBallotProtocol.abandonBallot(0);
 }
 
+void
+Slot::receivedTxSet(Value const& value)
+{
+    mBallotProtocol.receivedTxSet(value);
+}
+
 bool
 Slot::bumpState(Value const& value, bool force)
 {

src/scp/Slot.h

@@ -117,6 +117,10 @@ class Slot : public std::enable_shared_from_this<Slot>
 
     bool abandonBallot();
 
+    // Notify this slot that the tx set referenced by @p value has arrived so
+    // that it may resume balloting if stalled waiting for this tx set.
+    void receivedTxSet(Value const& value);
+
     // bumps the ballot based on the local state and the value passed in:
     // in prepare phase, attempts to take value
     // otherwise, no-ops