From 9febb4957efee90188141d65e0bbb04a92d1a364 Mon Sep 17 00:00:00 2001 From: Nicholas Dudfield Date: Tue, 14 Jul 2026 16:17:15 +0700 Subject: [PATCH] fix(export): make share snapshots terminal-safe --- src/test/rpc/Subscribe_test.cpp | 5 +- .../app/consensus/ConsensusExtensions.cpp | 107 +++++++++++++----- src/xrpld/app/consensus/ConsensusExtensions.h | 10 ++ src/xrpld/app/misc/NetworkOPs.cpp | 1 + src/xrpld/app/misc/NetworkOPs.h | 1 + 5 files changed, 95 insertions(+), 29 deletions(-) diff --git a/src/test/rpc/Subscribe_test.cpp b/src/test/rpc/Subscribe_test.cpp index f43c6d127..ce983b863 100644 --- a/src/test/rpc/Subscribe_test.cpp +++ b/src/test/rpc/Subscribe_test.cpp @@ -495,6 +495,7 @@ public: snapshot.validatedLedgerSeq && event[jss::ledger_hash] == to_string(snapshot.validatedLedgerHash) && + !event["terminal"].asBool() && event[jss::origin_txid] == to_string(share.originTxn) && shares.isArray() && shares.size() == 1 && shares[0u][jss::universe_position].asUInt() == @@ -508,12 +509,14 @@ public: snapshot.validatedLedgerSeq += 1; snapshot.validatedLedgerHash = uint256{5}; snapshot.shares.clear(); + snapshot.terminal = true; env.app().getOPs().pubExportSignatureSnapshot(snapshot); BEAST_EXPECT(wsc->findMsg(5s, [&](Json::Value const& event) { return event[jss::type] == "exportSignatureSnapshot" && event[jss::ledger_index].asUInt() == snapshot.validatedLedgerSeq && - event[jss::shares].isArray() && event[jss::shares].size() == 0; + event["terminal"].asBool() && event[jss::shares].isArray() && + event[jss::shares].size() == 0; })); jv = wsc->invoke("unsubscribe", stream); diff --git a/src/xrpld/app/consensus/ConsensusExtensions.cpp b/src/xrpld/app/consensus/ConsensusExtensions.cpp index cc9eeb7d6..905ce0ebc 100644 --- a/src/xrpld/app/consensus/ConsensusExtensions.cpp +++ b/src/xrpld/app/consensus/ConsensusExtensions.cpp @@ -74,6 +74,36 @@ struct ResolvedExportShare STTx releaseTarget; }; +bool +isPendingExportShare( + ReadView const& view, + ExportShare const& share, + beast::Journal j) +{ + if (share.originLedgerSeq > view.info().seq) + return false; + + auto const originHash = share.originLedgerSeq == view.info().seq + ? std::optional{view.info().hash} + : hashOfSeq(view, share.originLedgerSeq, j); + if (!originHash || *originHash != share.originLedgerHash) + return false; + + auto const latch = + view.read(keylet::shadowTicket(share.owner, share.originTxn)); + return latch && latch->getType() == ltSHADOW_TICKET && + latch->isFieldPresent(sfTransactionHash) && + latch->isFieldPresent(sfExportUniverseHash) && + latch->isFieldPresent(sfExportCommittee) && + latch->isFieldPresent(sfLastLedgerSequence) && + latch->isFieldPresent(sfExportNode) && + !latch->isFieldPresent(sfExportSignatureHash) && + latch->getAccountID(sfAccount) == share.owner && + latch->getFieldH256(sfTransactionHash) == share.originTxn && + latch->getFieldU32(sfLedgerSequence) == share.originLedgerSeq && + view.info().seq <= latch->getFieldU32(sfLastLedgerSequence); +} + std::optional resolveExportShare( Application& app, @@ -84,26 +114,12 @@ resolveExportShare( { if (!validated || !validated->rules().enabled(featureExport) || share.triggerTxn != share.originTxn || - share.originLedgerSeq > validated->info().seq) - return std::nullopt; - - auto const originHash = share.originLedgerSeq == validated->info().seq - ? std::optional{validated->info().hash} - : hashOfSeq(*validated, share.originLedgerSeq, j); - if (!originHash || *originHash != share.originLedgerHash) + !isPendingExportShare(*validated, share, j)) return std::nullopt; auto const latchKey = keylet::shadowTicket(share.owner, share.originTxn); auto const latch = validated->read(latchKey); - if (!latch || latch->getType() != ltSHADOW_TICKET || - !latch->isFieldPresent(sfTransactionHash) || - !latch->isFieldPresent(sfExportUniverseHash) || - !latch->isFieldPresent(sfExportCommittee) || - !latch->isFieldPresent(sfLastLedgerSequence) || - latch->getAccountID(sfAccount) != share.owner || - latch->getFieldH256(sfTransactionHash) != share.originTxn || - latch->getFieldU32(sfLedgerSequence) != share.originLedgerSeq || - validated->info().seq > latch->getFieldU32(sfLastLedgerSequence)) + if (!latch) return std::nullopt; auto const originLedger = @@ -251,6 +267,12 @@ ConsensusExtensions::onExportShare(ExportShare const& share) std::move(*admission.ticket), signatureVerified, validated->info().seq); if (outcome.result == ExportSigCollectorV2::AdmitResult::accepted) { + std::lock_guard streamLock(exportStreamMutex_); + if (!exportShareServiceStarted_.load(std::memory_order_acquire)) + return false; + auto const latest = app_.getLedgerMaster().getValidatedLedger(); + if (!latest || !isPendingExportShare(*latest, share, j_)) + return false; app_.getOPs().pubExportSignature(share); return true; } @@ -440,21 +462,18 @@ ConsensusExtensions::onValidatedLedger( app_.overlay().broadcast(message); } - auto observedSeq = - lastExportSnapshotSeq_.load(std::memory_order_relaxed); - while (observedSeq < validated->info().seq && - !lastExportSnapshotSeq_.compare_exchange_weak( - observedSeq, - validated->info().seq, - std::memory_order_acq_rel, - std::memory_order_relaxed)) - { - } - if (observedSeq < validated->info().seq) + std::lock_guard streamLock(exportStreamMutex_); + if (!exportShareServiceStarted_.load(std::memory_order_acquire)) + return; + if (lastExportSnapshotSeq_.load(std::memory_order_relaxed) < + validated->info().seq) { + lastExportSnapshotSeq_.store( + validated->info().seq, std::memory_order_relaxed); auto const live = pendingExportLatches(*validated); auto const unionSnapshot = postValidationExportSigCollector_.fullUnionSnapshot(); + std::map currentOrigins; for (auto const& [origin, latch] : live) { auto const originSeq = latch->getFieldU32(sfLedgerSequence); @@ -473,7 +492,8 @@ ConsensusExtensions::onValidatedLedger( originSeq, *originHash, origin, - {}}; + {}, + false}; if (auto const it = unionSnapshot.find(origin); it != unionSnapshot.end()) { @@ -492,8 +512,35 @@ ConsensusExtensions::onValidatedLedger( contribution.signature}); } } + currentOrigins.emplace( + origin, + ExportSnapshotOrigin{ + snapshot.version, + snapshot.owner, + snapshot.originLedgerSeq, + snapshot.originLedgerHash, + snapshot.triggerTxn}); app_.getOPs().pubExportSignatureSnapshot(snapshot); } + + for (auto const& [origin, prior] : publishedExportOrigins_) + { + if (currentOrigins.contains(origin)) + continue; + app_.getOPs().pubExportSignatureSnapshot( + ripple::ExportSignatureSnapshot{ + prior.version, + validated->info().seq, + validated->info().hash, + prior.owner, + origin, + prior.originLedgerSeq, + prior.originLedgerHash, + prior.triggerTxn, + {}, + true}); + } + publishedExportOrigins_ = std::move(currentOrigins); } } catch (std::exception const& e) @@ -511,6 +558,8 @@ ConsensusExtensions::onValidatedLedger( void ConsensusExtensions::startExportShareService() { + std::lock_guard streamLock(exportStreamMutex_); + publishedExportOrigins_.clear(); lastExportSnapshotSeq_.store(0, std::memory_order_relaxed); exportShareServiceStarted_.store(true, std::memory_order_release); } @@ -519,6 +568,8 @@ void ConsensusExtensions::stopExportShareService() noexcept { exportShareServiceStarted_.store(false, std::memory_order_release); + std::lock_guard streamLock(exportStreamMutex_); + publishedExportOrigins_.clear(); } //------------------------------------------------------------------------------ diff --git a/src/xrpld/app/consensus/ConsensusExtensions.h b/src/xrpld/app/consensus/ConsensusExtensions.h index 46db259ea..78a3e91aa 100644 --- a/src/xrpld/app/consensus/ConsensusExtensions.h +++ b/src/xrpld/app/consensus/ConsensusExtensions.h @@ -52,6 +52,16 @@ class ConsensusExtensions Application& app_; ExportSigCollectorV2 postValidationExportSigCollector_; + struct ExportSnapshotOrigin + { + std::uint8_t version; + AccountID owner; + LedgerIndex originLedgerSeq; + uint256 originLedgerHash; + uint256 triggerTxn; + }; + std::mutex exportStreamMutex_; + std::map publishedExportOrigins_; std::set> proposalPublishedExportShares_; std::atomic exportShareServiceStarted_{false}; diff --git a/src/xrpld/app/misc/NetworkOPs.cpp b/src/xrpld/app/misc/NetworkOPs.cpp index 3442c6304..9fc1e7f1d 100644 --- a/src/xrpld/app/misc/NetworkOPs.cpp +++ b/src/xrpld/app/misc/NetworkOPs.cpp @@ -2451,6 +2451,7 @@ NetworkOPsImp::pubExportSignatureSnapshot( event[jss::origin_ledger_seq] = Json::UInt(snapshot.originLedgerSeq); event[jss::origin_ledger_hash] = to_string(snapshot.originLedgerHash); event[jss::trigger_txid] = to_string(snapshot.triggerTxn); + event["terminal"] = snapshot.terminal; auto& shares = event[jss::shares] = Json::arrayValue; for (auto const& share : snapshot.shares) diff --git a/src/xrpld/app/misc/NetworkOPs.h b/src/xrpld/app/misc/NetworkOPs.h index 499263264..19d3741b5 100644 --- a/src/xrpld/app/misc/NetworkOPs.h +++ b/src/xrpld/app/misc/NetworkOPs.h @@ -58,6 +58,7 @@ struct ExportSignatureSnapshot uint256 originLedgerHash; uint256 triggerTxn; std::vector shares; + bool terminal{false}; }; /** Provides server functionality for clients.