Resume stalled balloting immediately on tx set arrival

src/herder/HerderSCPDriver.cpp

@@ -1322,6 +1322,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);
+        }
     }
 }
 
@@ -1340,6 +1348,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;
         }
     }
@@ -1712,6 +1741,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
@@ -1745,6 +1791,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

@@ -177,6 +177,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);
@@ -309,6 +313,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/herder/test/HerderTests.cpp

@@ -9048,6 +9048,120 @@ TEST_CASE_VERSIONS("Herder properly validates when tx set is missing",
         });
 }
 
+// Test that a stalled ballot resumes immediately on tx set arrival
+TEST_CASE_VERSIONS("tx set arrival resumes stalled balloting", "[herder]")
+{
+    Config cfg(getTestConfig());
+    cfg.MANUAL_CLOSE = false;
+    cfg.EXPERIMENTAL_PARALLEL_TX_SET_DOWNLOAD = true;
+
+    VirtualClock clock;
+
+    auto v1Key = SecretKey::pseudoRandomForTesting();
+    auto v2Key = SecretKey::pseudoRandomForTesting();
+    auto const& v1Pk = v1Key.getPublicKey();
+    auto const& v2Pk = v2Key.getPublicKey();
+
+    // Local quorum set {self, v1, v2} with threshold 2
+    cfg.QUORUM_SET.threshold = 2;
+    cfg.QUORUM_SET.validators.emplace_back(v1Pk);
+    cfg.QUORUM_SET.validators.emplace_back(v2Pk);
+
+    Application::pointer app = createTestApplication(clock, cfg);
+
+    for_versions_from(
+        static_cast<uint32_t>(EMPTY_TX_SET_PROTOCOL_VERSION), *app, [&] {
+            auto const lcl =
+                app->getLedgerManager().getLastClosedLedgerHeader();
+            uint64_t const slotIndex = lcl.header.ledgerSeq + 1;
+            auto& herder = dynamic_cast<HerderImpl&>(app->getHerder());
+            auto& pendingEnvelopes = herder.getPendingEnvelopes();
+
+            // Peers use the same 3-node qset; pre-cache it so envelopes don't
+            // block on a qset fetch.
+            SCPQuorumSet qSet;
+            qSet.threshold = 2;
+            qSet.validators.push_back(cfg.NODE_SEED.getPublicKey());
+            qSet.validators.push_back(v1Pk);
+            qSet.validators.push_back(v2Pk);
+            auto qSetHash = sha256(xdr::xdr_to_opaque(qSet));
+            pendingEnvelopes.addSCPQuorumSet(qSetHash, qSet);
+
+            // Create a non-empty tx set that the node does not have
+            auto root = app->getRoot();
+            std::vector<TransactionFrameBasePtr> txs = {
+                root->tx({payment(root->getPublicKey(), 1)})};
+            auto txSet = makeTxSetFromTransactions(txs, *app, 0, 0).first;
+            auto txSetHash = txSet->getContentsHash();
+
+            auto sv = herder.makeStellarValue(txSetHash,
+                                              lcl.header.scpValue.closeTime + 1,
+                                              emptyUpgradeSteps, v1Key);
+            auto opaqueValue = xdr::xdr_to_opaque(sv);
+
+            auto makePrepareFromPeer = [&](SecretKey const& peerKey) {
+                SCPEnvelope env;
+                env.statement.slotIndex = slotIndex;
+                env.statement.pledges.type(SCP_ST_PREPARE);
+                auto& prep = env.statement.pledges.prepare();
+                prep.ballot.counter = 1;
+                prep.ballot.value = opaqueValue;
+                prep.prepared.activate() = prep.ballot;
+                prep.quorumSetHash = qSetHash;
+                env.statement.nodeID = peerKey.getPublicKey();
+                herder.signEnvelope(peerKey, env);
+                return env;
+            };
+
+            // Both peers accept-prepared (1, v). The envelopes are
+            // ready without the tx set (parallel downloading), and processing
+            // them drives the local node to confirm-prepared and then stall
+            // because the tx set is still missing.
+            REQUIRE(herder.recvSCPEnvelope(makePrepareFromPeer(v1Key)) ==
+                    Herder::ENVELOPE_STATUS_READY);
+            REQUIRE(herder.recvSCPEnvelope(makePrepareFromPeer(v2Key)) ==
+                    Herder::ENVELOPE_STATUS_READY);
+
+            auto localPrepare = [&]() {
+                auto const* env = herder.getSCP().getLatestMessage(
+                    cfg.NODE_SEED.getPublicKey());
+                REQUIRE(env);
+                REQUIRE(env->statement.pledges.type() == SCP_ST_PREPARE);
+                return env->statement.pledges.prepare();
+            };
+
+            // Stalled: h is set but the commit is deferred (nC == 0).
+            {
+                auto const prep = localPrepare();
+                REQUIRE(prep.ballot.counter == 1);
+                REQUIRE(prep.nH == 1);
+                REQUIRE(prep.nC == 0);
+            }
+
+            // Deliver the tx set
+            REQUIRE(herder.recvTxSet(txSetHash, txSet));
+
+            // Resumed: the commit completed at the same counter, indicating
+            // the lack of a ballot timeout
+            {
+                auto const prep = localPrepare();
+                REQUIRE(prep.ballot.counter == 1);
+                REQUIRE(prep.nH == 1);
+                REQUIRE(prep.nC == 1);
+            }
+
+            // Repeat delivery is a no-op: the tx set is no longer being
+            // fetched, and balloting state does not change.
+            REQUIRE(!herder.recvTxSet(txSetHash, txSet));
+            {
+                auto const prep = localPrepare();
+                REQUIRE(prep.ballot.counter == 1);
+                REQUIRE(prep.nH == 1);
+                REQUIRE(prep.nC == 1);
+            }
+        });
+}
+
 #ifdef CAP_0083
 // This tests that the network externalizes an empty-tx-set value when a
 // voted-for value is not available on the network.

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);
+                        }
                     }
                 }
             }
@@ -675,7 +684,7 @@ BallotProtocol::createStatement(SCPStatementType const& type)
     return statement;
 }
 
-void
+SCPStatement
 BallotProtocol::emitCurrentStateStatement()
 {
     ZoneScoped;
@@ -729,6 +738,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
@@ -1144,6 +1158,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);
@@ -1179,6 +1194,8 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH)
                 mSlot.getSCPDriver().recordBallotBlockedOnTxSet(
                     mSlot.getSlotIndex(), newC.value);
 
+                stalled = true;
+
                 CLOG_TRACE(
                     SCP,
                     "BallotProtocol::setConfirmPrepared slot:{} "
@@ -1217,12 +1234,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,