Implement subscription for book_changes (#315)

Fixes #315

src/backend/CassandraBackend.h

@@ -1026,13 +1026,13 @@ public:
     isTooBusy() const override;
 
     inline void
-    incremementOutstandingRequestCount() const
+    incrementOutstandingRequestCount() const
     {
         {
             std::unique_lock<std::mutex> lck(throttleMutex_);
             if (!canAddRequest())
             {
-                BOOST_LOG_TRIVIAL(info)
+                BOOST_LOG_TRIVIAL(debug)
                     << __func__ << " : "
                     << "Max outstanding requests reached. "
                     << "Waiting for other requests to finish";
@@ -1109,7 +1109,7 @@ public:
         bool isRetry) const
     {
         if (!isRetry)
-            incremementOutstandingRequestCount();
+            incrementOutstandingRequestCount();
         executeAsyncHelper(statement, callback, callbackData);
     }
 

src/etl/ReportingETL.cpp

@@ -202,6 +202,9 @@ ReportingETL::publishLedger(ripple::LedgerInfo const& lgrInfo)
 
         for (auto& txAndMeta : transactions)
             subscriptions_->pubTransaction(txAndMeta, lgrInfo);
+
+        subscriptions_->pubBookChanges(lgrInfo, transactions);
+
         BOOST_LOG_TRIVIAL(info) << __func__ << " - Published ledger "
                                 << std::to_string(lgrInfo.seq);
     }

src/rpc/Handlers.h

@@ -45,7 +45,7 @@ Result
 doChannelVerify(Context const& context);
 
 // book methods
-Result
+[[nodiscard]] Result
 doBookChanges(Context const& context);
 
 Result

src/rpc/RPC.h

@@ -9,6 +9,7 @@
 #include <rpc/Counters.h>
 #include <string>
 #include <variant>
+
 /*
  * This file contains various classes necessary for executing RPC handlers.
  * Context gives the handlers access to various other parts of the application

src/rpc/RPCHelpers.h

@@ -270,5 +270,10 @@ traverseTransactions(
         std::optional<Backend::TransactionsCursor> const&,
         boost::asio::yield_context& yield)> transactionFetcher);
 
+[[nodiscard]] boost::json::object const
+computeBookChanges(
+    ripple::LedgerInfo const& lgrInfo,
+    std::vector<Backend::TransactionAndMetadata> const& transactions);
+
 }  // namespace RPC
 #endif

src/rpc/handlers/BookChanges.cpp

@@ -12,6 +12,9 @@ using namespace ripple;
 
 namespace RPC {
 
+/**
+ * @brief Represents an entry in the book_changes' changes array.
+ */
 struct BookChange
 {
     STAmount sideAVolume;
@@ -22,44 +25,39 @@ struct BookChange
     STAmount closeRate;
 };
 
-class BookChangesHandler
+/**
+ * @brief Encapsulates the book_changes computations and transformations.
+ */
+class BookChanges final
+{
+public:
+    BookChanges() = delete;  // only accessed via static handle function
+
+    /**
+     * @brief Computes all book_changes for the given transactions.
+     *
+     * @param transactions The transactions to compute book changes for
+     * @return std::vector<BookChange> Book changes
+     */
+    [[nodiscard]] static std::vector<BookChange>
+    compute(std::vector<Backend::TransactionAndMetadata> const& transactions)
+    {
+        return HandlerImpl{}(transactions);
+    }
+
+private:
+    class HandlerImpl final
     {
-    std::reference_wrapper<Context const> context_;
         std::map<std::string, BookChange> tally_ = {};
         std::optional<uint32_t> offerCancel_ = {};
 
     public:
-    ~BookChangesHandler() = default;
+        [[nodiscard]] std::vector<BookChange>
-    explicit BookChangesHandler(Context const& context)
+        operator()(
-        : context_{std::cref(context)}
+            std::vector<Backend::TransactionAndMetadata> const& transactions)
-    {
-    }
-
-    BookChangesHandler(BookChangesHandler const&) = delete;
-    BookChangesHandler(BookChangesHandler&&) = delete;
-    BookChangesHandler&
-    operator=(BookChangesHandler const&) = delete;
-    BookChangesHandler&
-    operator=(BookChangesHandler&&) = delete;
-
-    /**
-     * @brief Handles the `book_change` request for given transactions
-     *
-     * @param transactions The transactions to compute changes for
-     * @return std::vector<BookChange> The changes
-     */
-    std::vector<BookChange>
-    handle(LedgerInfo const& ledger)
-    {
-        reset();
-
-        for (auto const transactions =
-                 context_.get().backend->fetchAllTransactionsInLedger(
-                     ledger.seq, context_.get().yield);
-             auto const& tx : transactions)
         {
+            for (auto const& tx : transactions)
                 handleBookChange(tx);
-        }
 
             // TODO: rewrite this with std::ranges when compilers catch up
             std::vector<BookChange> changes;
@@ -72,13 +70,6 @@ public:
         }
 
     private:
-    inline void
-    reset() noexcept
-    {
-        tally_.clear();
-        offerCancel_ = std::nullopt;
-    }
-
         void
         handleAffectedNode(STObject const& node)
         {
@@ -121,6 +112,14 @@ private:
             auto const deltaPays = finalFields.getFieldAmount(sfTakerPays) -
                 previousFields.getFieldAmount(sfTakerPays);
 
+            transformAndStore(deltaGets, deltaPays);
+        }
+
+        void
+        transformAndStore(
+            ripple::STAmount const& deltaGets,
+            ripple::STAmount const& deltaPays)
+        {
             auto const g = to_string(deltaGets.issue());
             auto const p = to_string(deltaPays.issue());
 
@@ -200,6 +199,7 @@ private:
             }
         }
     };
+};
 
 void
 tag_invoke(
@@ -228,6 +228,20 @@ tag_invoke(
     };
 }
 
+json::object const
+computeBookChanges(
+    ripple::LedgerInfo const& lgrInfo,
+    std::vector<Backend::TransactionAndMetadata> const& transactions)
+{
+    return {
+        {JS(type), "bookChanges"},
+        {JS(ledger_index), lgrInfo.seq},
+        {JS(ledger_hash), to_string(lgrInfo.hash)},
+        {JS(ledger_time), lgrInfo.closeTime.time_since_epoch().count()},
+        {JS(changes), json::value_from(BookChanges::compute(transactions))},
+    };
+}
+
 Result
 doBookChanges(Context const& context)
 {
@@ -237,14 +251,9 @@ doBookChanges(Context const& context)
         return *status;
 
     auto const lgrInfo = std::get<ripple::LedgerInfo>(info);
-    auto const changes = BookChangesHandler{context}.handle(lgrInfo);
+    auto const transactions = context.backend->fetchAllTransactionsInLedger(
-    return json::object{
+        lgrInfo.seq, context.yield);
-        {JS(type), "bookChanges"},
+    return computeBookChanges(lgrInfo, transactions);
-        {JS(ledger_index), lgrInfo.seq},
-        {JS(ledger_hash), to_string(lgrInfo.hash)},
-        {JS(ledger_time), lgrInfo.closeTime.time_since_epoch().count()},
-        {JS(changes), json::value_from(changes)},
-    };
 }
 
 }  // namespace RPC

src/rpc/handlers/Subscribe.cpp

@@ -12,7 +12,8 @@ static std::unordered_set<std::string> validCommonStreams{
     "transactions",
     "transactions_proposed",
     "validations",
-    "manifests"};
+    "manifests",
+    "book_changes"};
 
 Status
 validateStreams(boost::json::object const& request)
@@ -57,6 +58,8 @@ subscribeToStreams(
             manager.subValidation(session);
         else if (s == "manifests")
             manager.subManifest(session);
+        else if (s == "book_changes")
+            manager.subBookChanges(session);
         else
             assert(false);
     }
@@ -85,6 +88,8 @@ unsubscribeToStreams(
             manager.unsubValidation(session);
         else if (s == "manifests")
             manager.unsubManifest(session);
+        else if (s == "book_changes")
+            manager.unsubBookChanges(session);
         else
             assert(false);
     }

src/subscriptions/SubscriptionManager.cpp

@@ -70,7 +70,7 @@ Subscription::unsubscribe(std::shared_ptr<WsBase> const& session)
 }
 
 void
-Subscription::publish(std::shared_ptr<Message>& message)
+Subscription::publish(std::shared_ptr<Message> const& message)
 {
     boost::asio::post(strand_, [this, message]() {
         sendToSubscribers(message, subscribers_, subCount_);
@@ -235,6 +235,22 @@ SubscriptionManager::unsubBook(
     bookSubscribers_.unsubscribe(session, book);
 }
 
+void
+SubscriptionManager::subBookChanges(std::shared_ptr<WsBase> session)
+{
+    bookChangesSubscribers_.subscribe(session);
+
+    std::unique_lock lk(cleanupMtx_);
+    cleanupFuncs_[session].emplace_back(
+        [this](session_ptr session) { unsubBookChanges(session); });
+}
+
+void
+SubscriptionManager::unsubBookChanges(std::shared_ptr<WsBase> session)
+{
+    bookChangesSubscribers_.unsubscribe(session);
+}
+
 void
 SubscriptionManager::pubLedger(
     ripple::LedgerInfo const& lgrInfo,
@@ -345,6 +361,20 @@ SubscriptionManager::pubTransaction(
     }
 }
 
+void
+SubscriptionManager::pubBookChanges(
+    ripple::LedgerInfo const& lgrInfo,
+    std::vector<Backend::TransactionAndMetadata> const& transactions)
+{
+    if (bookChangesSubscribers_.empty())
+        return;
+
+    auto const json = RPC::computeBookChanges(lgrInfo, transactions);
+    auto const bookChangesMsg =
+        std::make_shared<Message>(boost::json::serialize(json));
+    bookChangesSubscribers_.publish(bookChangesMsg);
+}
+
 void
 SubscriptionManager::forwardProposedTransaction(
     boost::json::object const& response)

src/subscriptions/SubscriptionManager.h

@@ -31,13 +31,19 @@ public:
     unsubscribe(std::shared_ptr<WsBase> const& session);
 
     void
-    publish(std::shared_ptr<Message>& message);
+    publish(std::shared_ptr<Message> const& message);
 
     std::uint64_t
-    count()
+    count() const
     {
         return subCount_.load();
     }
+
+    bool
+    empty() const
+    {
+        return count() == 0;
+    }
 };
 
 template <class Key>
@@ -90,6 +96,7 @@ class SubscriptionManager
     Subscription txProposedSubscribers_;
     Subscription manifestSubscribers_;
     Subscription validationsSubscribers_;
+    Subscription bookChangesSubscribers_;
 
     SubscriptionMap<ripple::AccountID> accountSubscribers_;
     SubscriptionMap<ripple::AccountID> accountProposedSubscribers_;
@@ -122,6 +129,7 @@ public:
         , txProposedSubscribers_(ioc_)
         , manifestSubscribers_(ioc_)
         , validationsSubscribers_(ioc_)
+        , bookChangesSubscribers_(ioc_)
         , accountSubscribers_(ioc_)
         , accountProposedSubscribers_(ioc_)
         , bookSubscribers_(ioc_)
@@ -159,6 +167,11 @@ public:
         std::string const& ledgerRange,
         std::uint32_t txnCount);
 
+    void
+    pubBookChanges(
+        ripple::LedgerInfo const& lgrInfo,
+        std::vector<Backend::TransactionAndMetadata> const& transactions);
+
     void
     unsubLedger(session_ptr session);
 
@@ -185,6 +198,12 @@ public:
     void
     unsubBook(ripple::Book const& book, session_ptr session);
 
+    void
+    subBookChanges(std::shared_ptr<WsBase> session);
+
+    void
+    unsubBookChanges(std::shared_ptr<WsBase> session);
+
     void
     subManifest(session_ptr session);
 
@@ -234,6 +253,7 @@ public:
         counts["account"] = accountSubscribers_.count();
         counts["accounts_proposed"] = accountProposedSubscribers_.count();
         counts["books"] = bookSubscribers_.count();
+        counts["book_changes"] = bookChangesSubscribers_.count();
 
         return counts;
     }

test.py

@@ -836,6 +836,7 @@ async def subscribe(ip, port):
     try:
         async with websockets.connect(address) as ws:
             await ws.send(json.dumps({"command":"subscribe","streams":["ledger"]}))
+            #await ws.send(json.dumps({"command":"subscribe","streams":["book_changes"]}))
             #await ws.send(json.dumps({"command":"subscribe","streams":["manifests"]}))
             while True:
                 res = json.loads(await ws.recv())