RXI-1136 allow multi commits with the same claimID

oleks-rip · Jun 15, 2024 · a03d3a0 · a03d3a0
1 parent e70d832
commit a03d3a0
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 12 deletions.
diff --git a/src/xbwd/federator/Federator.cpp b/src/xbwd/federator/Federator.cpp
@@ -347,7 +347,7 @@ Federator::readDBAttests(ChainType ct)
             }
 
             txnsInProcessing_[ct].insert(
-                fmt::format("create: {:x}", createCount));
+                {fmt::format("create: {:x}", createCount), transID});
             if (!autoSubmit_[ct])
                 continue;
 
@@ -458,7 +458,8 @@ Federator::readDBAttests(ChainType ct)
                 continue;
             }
 
-            txnsInProcessing_[ct].insert(fmt::format("claim: {:x}", claimID));
+            txnsInProcessing_[ct].insert(
+                {fmt::format("claim: {:x}", claimID), transID});
             if (!autoSubmit_[ct])
                 continue;
 
@@ -781,17 +782,15 @@ Federator::onEvent(event::XChainCommitDetected const& e)
     }
 
     auto const txnIdHex = ripple::strHex(e.txnHash_.begin(), e.txnHash_.end());
-    auto const res =
-        txnsInProcessing_[ct].insert(fmt::format("claim: {:x}", e.claimID_));
+    auto const res = txnsInProcessing_[ct].insert(
+        {fmt::format("claim: {:x}", e.claimID_), txnIdHex});
     if (!res.second)
     {
         // Already have this transaction
         // TODO: Sanity check the claim id and deliveredAmt match
         // TODO: Stop historical transaction collection
         JLOGV(
-            j_.error(),
-            "XChainCommit already present",
-            jv("event", e.toJson()));
+            j_.warn(), "XChainCommit already present", jv("event", e.toJson()));
         return;  // Don't store it again
     }
 
@@ -991,7 +990,7 @@ Federator::onEvent(event::XChainAccountCreateCommitDetected const& e)
 
     auto const txnIdHex = ripple::strHex(e.txnHash_.begin(), e.txnHash_.end());
     auto const res = txnsInProcessing_[ct].insert(
-        fmt::format("create: {:x}", e.createCount_));
+        {fmt::format("create: {:x}", e.createCount_), txnIdHex});
     if (!res.second)
     {
         // Already have this transaction
@@ -1331,15 +1330,15 @@ Federator::onEvent(event::XChainAttestsResult const& e)
         if ((e.type_ == xbwd::XChainTxnType::xChainAddClaimAttestation) &&
             e.claimID_)
         {
-            cnt = txnsInProcessing_[oct].erase(
+            cnt = txnsInProcessing_[oct].get<1>().erase(
                 fmt::format("claim: {:x}", *e.claimID_));
         }
         else if (
             (e.type_ ==
              xbwd::XChainTxnType::xChainAddAccountCreateAttestation) &&
             e.createCount_)
         {
-            cnt = txnsInProcessing_[oct].erase(
+            cnt = txnsInProcessing_[oct].get<1>().erase(
                 fmt::format("create: {:x}", *e.createCount_));
         }
         else

diff --git a/src/xbwd/federator/Federator.h b/src/xbwd/federator/Federator.h
@@ -36,6 +36,10 @@
 #include <ripple/protocol/XChainAttestations.h>
 
 #include <boost/asio.hpp>
+#include <boost/multi_index/hashed_index.hpp>
+#include <boost/multi_index/identity.hpp>
+#include <boost/multi_index/member.hpp>
+#include <boost/multi_index_container.hpp>
 
 #include <atomic>
 #include <condition_variable>
@@ -289,6 +293,23 @@ struct AttestedHistoryTx
     fromEvent(FederatorEvent const& e);
 };
 
+struct TransactionCache
+{
+    std::string id_;  // ClaimID or CreateID
+    std::string tx_;  // tx hash of XChainCommit /  XChainAccountCreateCommit
+                      // transaction
+};
+
+namespace mi = boost::multi_index;
+typedef mi::multi_index_container<
+    TransactionCache,
+    mi::indexed_by<
+        mi::hashed_unique<
+            mi::member<TransactionCache, std::string, &TransactionCache::tx_>>,
+        mi::hashed_non_unique<
+            mi::member<TransactionCache, std::string, &TransactionCache::id_>>>>
+    TransactionCacheContainer;
+
 class Federator
 {
     enum LoopTypes {
@@ -336,8 +357,10 @@ class Federator
     ChainArray<std::vector<SubmissionPtr>> GUARDED_BY(txnsMutex_) errored_;
 
     // Cache of the events added to processing. It is added so as not to read
-    // the DB. No need for mutex as event processing is synchronized.
-    ChainArray<std::unordered_set<std::string>> txnsInProcessing_;
+    // the DB. No need for mutex as event processing is synchronized. Insertion
+    // check transaction id for uniqueness. Deletion remove all the entry with
+    // the same claimID
+    ChainArray<TransactionCacheContainer> txnsInProcessing_;
 
     // "Window" size for sending attestations
     // 0 - no "window"