fix: Bound and offload per-connection subscription cleanup

src/libxrpl/server/InfoSub.cpp

@@ -7,10 +7,12 @@
 #include <xrpl/protocol/Book.h>
 #include <xrpl/resource/Consumer.h>
 
+#include <cstddef>
 #include <cstdint>
 #include <exception>
 #include <memory>
 #include <mutex>
+#include <utility>
 
 namespace xrpl {
 
@@ -64,6 +66,9 @@ InfoSub::InfoSub(Source& source, Consumer consumer)
 
 InfoSub::~InfoSub()
 {
+    // Stream unsubscribes are O(1): each erases this connection's single seq_
+    // from one stream map, so they are cheap enough to run inline on the
+    // disconnect thread.
     // Each Source teardown call below acquires a server-side lock and
     // can throw. Wrap each independent call so partial failure does not
     // skip the remaining teardown steps.
@@ -79,29 +84,48 @@ InfoSub::~InfoSub()
     safeUnsub(seq_, [&] { source_.unsubPeerStatus(seq_); }, j);
     safeUnsub(seq_, [&] { source_.unsubConsensus(seq_); }, j);
 
-    // Use the internal unsubscribe so that it won't call
-    // back to us and modify its own parameter
-    if (!realTimeSubscriptions_.empty())
-    {
-        safeUnsub(
-            seq_, [&] { source_.unsubAccountInternal(seq_, realTimeSubscriptions_, true); }, j);
-    }
-
-    if (!normalSubscriptions_.empty())
-    {
-        safeUnsub(
-            seq_, [&] { source_.unsubAccountInternal(seq_, normalSubscriptions_, false); }, j);
-    }
-
-    for (auto const& account : accountHistorySubscriptions_)
-    {
-        safeUnsub(seq_, [&] { source_.unsubAccountHistoryInternal(seq_, account, false); }, j);
-    }
-
+    // Book subscriptions are torn down inline here, keyed on seq_, rather than
+    // through the chunked account cleanup below. The book set is not capped, so
+    // it can be large; but each unsubBookInternal takes bookLock_ for a single
+    // O(1) erase and releases it, so even a large set never holds a lock across
+    // the whole loop - a competing book publish can interleave between erases.
+    // The disconnect thread still does O(N) brief acquisitions. Use the internal
+    // variant so it does not write back to bookSubscriptions_ on this
+    // partially-destroyed object.
     for (auto const& book : bookSubscriptions_)
     {
         safeUnsub(seq_, [&] { source_.unsubBookInternal(seq_, book); }, j);
     }
+
+    // Hand the account sets off (by move) to the Source for a chunked,
+    // off-thread teardown keyed on seq_, instead of erasing them inline here.
+    // This keeps the destructor from holding the account lock across a large
+    // erase loop. The job never references this object, which is being
+    // destroyed.
+    //
+    // Moving the sets without holding lock_ is safe: the destructor runs only
+    // when the last shared_ptr to this InfoSub is released, so by the
+    // shared_ptr contract no other thread holds a reference. Subscription maps
+    // store weak_ptrs, so a concurrent publisher must weak_ptr::lock() first;
+    // that succeeds only while a strong reference exists, which cannot overlap
+    // with destruction. No other thread can observe the moved-from sets.
+    //
+    // Wrapped like the steps above: scheduleAccountCleanup enqueues a JobQueue
+    // task, which allocates and locks and so can throw. A throw out of this
+    // noexcept destructor would terminate the process. Skipping the cleanup on
+    // throw is harmless: the account/rt maps hold weak_ptrs that the next
+    // publish prunes once this InfoSub is gone, and any history paging job
+    // self-terminates when its weak sink can no longer be locked.
+    safeUnsub(
+        seq_,
+        [&] {
+            source_.scheduleAccountCleanup(
+                seq_,
+                std::move(realTimeSubscriptions_),
+                std::move(normalSubscriptions_),
+                std::move(accountHistorySubscriptions_));
+        },
+        j);
 }
 
 Resource::Consumer&
@@ -121,6 +145,53 @@ InfoSub::onSendEmpty()
 {
 }
 
+std::size_t
+InfoSub::totalSubscriptionCount() const
+{
+    // Hold lock_ for the whole read so the three sets cannot be mutated
+    // mid-count by a concurrent (un)subscribe on this connection.
+    std::scoped_lock const sl(lock_);
+
+    // Combined tally the per-connection cap is enforced against.
+    return normalSubscriptions_.size() + realTimeSubscriptions_.size() +
+        accountHistorySubscriptions_.size();
+}
+
+bool
+InfoSub::tryReserveAccountSubscriptions(
+    hash_set<AccountID> const& proposedAccounts,
+    hash_set<AccountID> const& normalAccounts,
+    std::size_t cap)
+{
+    // One lock hold covers the count, the check and the insert.
+    std::scoped_lock const sl(lock_);
+
+    // Entries not already tracked; re-subscribing held accounts is not charged.
+    auto const countNew = [](hash_set<AccountID> const& requested,
+                             hash_set<AccountID> const& existing) {
+        std::size_t fresh = 0;
+        for (auto const& account : requested)
+        {
+            if (!existing.contains(account))
+                ++fresh;
+        }
+        return fresh;
+    };
+
+    std::size_t const additional = countNew(proposedAccounts, realTimeSubscriptions_) +
+        countNew(normalAccounts, normalSubscriptions_);
+
+    std::size_t const current = normalSubscriptions_.size() + realTimeSubscriptions_.size() +
+        accountHistorySubscriptions_.size();
+
+    if (exceedsSubscriptionCap(current, additional, cap))
+        return false;
+
+    realTimeSubscriptions_.insert(proposedAccounts.begin(), proposedAccounts.end());
+    normalSubscriptions_.insert(normalAccounts.begin(), normalAccounts.end());
+    return true;
+}
+
 void
 InfoSub::insertSubAccountInfo(AccountID const& account, bool rt)
 {
@@ -165,6 +236,13 @@ InfoSub::deleteSubAccountHistory(AccountID const& account)
     accountHistorySubscriptions_.erase(account);
 }
 
+bool
+InfoSub::hasAccountHistorySubscription(AccountID const& account) const
+{
+    std::scoped_lock const sl(lock_);
+    return accountHistorySubscriptions_.contains(account);
+}
+
 void
 InfoSub::insertBookSubscription(Book const& book)
 {