Fix: Duplicate messages when subscribe both accounts and proposed_accounts (#1415)

Fix #1134

src/etl/impl/SubscriptionSource.cpp

@@ -240,7 +240,11 @@ SubscriptionSource::handleMessage(std::string const& message)
 
         } else {
             if (isForwarding_) {
-                if (object.contains(JS(transaction))) {
+                // Clio as rippled's proposed_transactions subscirber, will receive two jsons for each transaction
+                // 1 - Proposed transaction
+                // 2 - Validated transaction
+                // Only forward proposed transaction, validated transactions are sent by Clio itself
+                if (object.contains(JS(transaction)) and !object.contains(JS(meta))) {
                     subscriptions_->forwardProposedTransaction(object);
                 } else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ValidationReceived) {
                     subscriptions_->forwardValidation(object);

src/feed/SubscriptionManager.cpp

@@ -59,24 +59,32 @@ void
 SubscriptionManager::subProposedTransactions(SubscriberSharedPtr const& subscriber)
 {
     proposedTransactionFeed_.sub(subscriber);
+    // proposed_transactions subscribers not only receive the transaction json when it is proposed, but also the
+    // transaction json when it is validated. So the subscriber also subscribes to the transaction feed.
+    transactionFeed_.subProposed(subscriber);
 }
 
 void
 SubscriptionManager::unsubProposedTransactions(SubscriberSharedPtr const& subscriber)
 {
     proposedTransactionFeed_.unsub(subscriber);
+    transactionFeed_.unsubProposed(subscriber);
 }
 
 void
 SubscriptionManager::subProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber)
 {
     proposedTransactionFeed_.sub(account, subscriber);
+    // Same as proposed_transactions subscribers, proposed_account subscribers also subscribe to the transaction feed to
+    // receive validated transaction feed. TransactionFeed class will filter out the sessions that have been sent to.
+    transactionFeed_.subProposed(account, subscriber);
 }
 
 void
 SubscriptionManager::unsubProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber)
 {
     proposedTransactionFeed_.unsub(account, subscriber);
+    transactionFeed_.unsubProposed(account, subscriber);
 }
 
 void

src/feed/impl/ProposedTransactionFeed.cpp

@@ -102,6 +102,7 @@ ProposedTransactionFeed::pub(boost::json::object const& receivedTxJson)
     auto affectedAccounts = std::unordered_set<ripple::AccountID>(accounts.cbegin(), accounts.cend());
 
     boost::asio::post(strand_, [this, pubMsg = std::move(pubMsg), affectedAccounts = std::move(affectedAccounts)]() {
+        notified_.clear();
         signal_.emit(pubMsg);
         // Prevent the same connection from receiving the same message twice if it is subscribed to multiple accounts
         // However, if the same connection subscribe both stream and account, it will still receive the message twice.

src/feed/impl/TransactionFeed.cpp

@@ -93,6 +93,27 @@ TransactionFeed::sub(ripple::AccountID const& account, SubscriberSharedPtr const
     }
 }
 
+void
+TransactionFeed::subProposed(SubscriberSharedPtr const& subscriber)
+{
+    auto const added = txProposedsignal_.connectTrackableSlot(subscriber, TransactionSlot(*this, subscriber));
+    if (added) {
+        subscriber->onDisconnect.connect([this](SubscriberPtr connection) { unsubProposedInternal(connection); });
+    }
+}
+
+void
+TransactionFeed::subProposed(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber)
+{
+    auto const added =
+        accountProposedSignal_.connectTrackableSlot(subscriber, account, TransactionSlot(*this, subscriber));
+    if (added) {
+        subscriber->onDisconnect.connect([this, account](SubscriberPtr connection) {
+            unsubProposedInternal(account, connection);
+        });
+    }
+}
+
 void
 TransactionFeed::sub(ripple::Book const& book, SubscriberSharedPtr const& subscriber)
 {
@@ -116,6 +137,18 @@ TransactionFeed::unsub(ripple::AccountID const& account, SubscriberSharedPtr con
     unsubInternal(account, subscriber.get());
 }
 
+void
+TransactionFeed::unsubProposed(SubscriberSharedPtr const& subscriber)
+{
+    unsubProposedInternal(subscriber.get());
+}
+
+void
+TransactionFeed::unsubProposed(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber)
+{
+    unsubProposedInternal(account, subscriber.get());
+}
+
 void
 TransactionFeed::unsub(ripple::Book const& book, SubscriberSharedPtr const& subscriber)
 {
@@ -251,14 +284,19 @@ TransactionFeed::pub(
          affectedBooks = std::move(affectedBooks)]() {
             notified_.clear();
             signal_.emit(allVersionsMsgs);
+            // clear the notified set. If the same connection subscribes both transactions + proposed_transactions,
+            // rippled SENDS the same message twice
             notified_.clear();
-            // check duplicate for accounts, this prevents sending the same message multiple times if it touches
-            // multiple accounts watched by the same connection
+            txProposedsignal_.emit(allVersionsMsgs);
+            notified_.clear();
+            // check duplicate for account and proposed_account, this prevents sending the same message multiple times
+            // if it affects multiple accounts watched by the same connection
             for (auto const& account : affectedAccounts) {
                 accountSignal_.emit(account, allVersionsMsgs);
+                accountProposedSignal_.emit(account, allVersionsMsgs);
             }
             notified_.clear();
-            // check duplicate for books, this prevents sending the same message multiple times if it touches multiple
+            // check duplicate for books, this prevents sending the same message multiple times if it affects multiple
             // books watched by the same connection
             for (auto const& book : affectedBooks) {
                 bookSignal_.emit(book, allVersionsMsgs);
@@ -285,6 +323,18 @@ TransactionFeed::unsubInternal(ripple::AccountID const& account, SubscriberPtr s
     }
 }
 
+void
+TransactionFeed::unsubProposedInternal(SubscriberPtr subscriber)
+{
+    txProposedsignal_.disconnect(subscriber);
+}
+
+void
+TransactionFeed::unsubProposedInternal(ripple::AccountID const& account, SubscriberPtr subscriber)
+{
+    accountProposedSignal_.disconnect(subscriber, account);
+}
+
 void
 TransactionFeed::unsubInternal(ripple::Book const& book, SubscriberPtr subscriber)
 {

src/feed/impl/TransactionFeed.hpp

@@ -72,6 +72,10 @@ class TransactionFeed {
     TrackableSignalMap<ripple::Book, Subscriber, AllVersionTransactionsType const&> bookSignal_;
     TrackableSignal<Subscriber, AllVersionTransactionsType const&> signal_;
 
+    // Signals for proposed tx subscribers
+    TrackableSignalMap<ripple::AccountID, Subscriber, AllVersionTransactionsType const&> accountProposedSignal_;
+    TrackableSignal<Subscriber, AllVersionTransactionsType const&> txProposedsignal_;
+
     std::unordered_set<SubscriberPtr>
         notified_;  // Used by slots to prevent double notifications if tx contains multiple subscribed accounts
 
@@ -111,6 +115,22 @@ public:
     void
     sub(ripple::Book const& book, SubscriberSharedPtr const& subscriber);
 
+    /**
+     * @brief Subscribe to the transaction feed for proposed transaction stream.
+     * @param subscriber
+     */
+    void
+    subProposed(SubscriberSharedPtr const& subscriber);
+
+    /**
+     * @brief Subscribe to the transaction feed for proposed account, only receive the feed when particular account is
+     * affected.
+     * @param subscriber
+     * @param account The account to watch.
+     */
+    void
+    subProposed(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
+
     /**
      * @brief Unsubscribe to the transaction feed.
      * @param subscriber
@@ -126,6 +146,21 @@ public:
     void
     unsub(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
 
+    /**
+     * @brief Unsubscribe to the transaction feed for proposed transaction stream.
+     * @param subscriber
+     */
+    void
+    unsubProposed(SubscriberSharedPtr const& subscriber);
+
+    /**
+     * @brief Unsubscribe to the transaction for particular proposed account.
+     * @param subscriber
+     * @param account The account to unsubscribe.
+     */
+    void
+    unsubProposed(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
+
     /**
      * @brief Unsubscribe to the transaction feed for particular order book.
      * @param subscriber
@@ -170,6 +205,12 @@ private:
     void
     unsubInternal(ripple::AccountID const& account, SubscriberPtr subscriber);
 
+    void
+    unsubProposedInternal(SubscriberPtr subscriber);
+
+    void
+    unsubProposedInternal(ripple::AccountID const& account, SubscriberPtr subscriber);
+
     void
     unsubInternal(ripple::Book const& book, SubscriberPtr subscriber);
 };