diff --git a/cfg/xrpld-example.cfg b/cfg/xrpld-example.cfg index 9e334e6f4f..6a44561c68 100644 --- a/cfg/xrpld-example.cfg +++ b/cfg/xrpld-example.cfg @@ -488,6 +488,17 @@ # Must be a number between 100 and 1000, defaults to 250 # # +# [max_subscriptions_per_connection] +# +# Maximum number of account, real-time account, and account-history +# subscriptions a single client connection may hold at once. Bounds the +# per-connection state torn down when the connection disconnects. Book +# subscriptions are tracked separately and are not counted here. +# +# Defaults to 100000 if not set; large enough for legitimate power users +# such as block explorers. +# +# # [overlay] # # Controls settings related to the peer to peer overlay. diff --git a/include/xrpl/config/Constants.h b/include/xrpl/config/Constants.h index 5514e0e77b..0fe4efee63 100644 --- a/include/xrpl/config/Constants.h +++ b/include/xrpl/config/Constants.h @@ -25,6 +25,7 @@ struct Sections static constexpr auto kLedgerHistory = "ledger_history"; static constexpr auto kLedgerReplay = "ledger_replay"; static constexpr auto kLedgerTxTables = "ledger_tx_tables"; + static constexpr auto kMaxSubscriptionsPerConnection = "max_subscriptions_per_connection"; static constexpr auto kMaxTransactions = "max_transactions"; static constexpr auto kNetworkId = "network_id"; static constexpr auto kNetworkQuorum = "network_quorum"; diff --git a/include/xrpl/server/InfoSub.h b/include/xrpl/server/InfoSub.h index 2e9bd857c7..1c12d9e520 100644 --- a/include/xrpl/server/InfoSub.h +++ b/include/xrpl/server/InfoSub.h @@ -11,6 +11,7 @@ #include #include +#include #include #include #include @@ -22,6 +23,39 @@ namespace xrpl { // Operations that clients may wish to perform against the network // Master operational handler, server sequencer, network tracker +/** + * Maximum number of subscriptions a single client connection may hold at once. + * + * Applies to the account, real-time account, and account-history subscriptions + * tracked on one InfoSub (the sets counted by totalSubscriptionCount), bounding + * the disconnect-time cleanup of those sets. Book subscriptions are tracked + * separately (OrderBookDB) and are not counted here. Generous enough for + * legitimate power users such as block explorers. + */ +constexpr std::size_t kMaxSubscriptionsPerConnection = 100'000; + +/** + * Whether adding @p additional subscriptions to a connection already holding + * @p current would exceed the cap. + * + * Pure arithmetic split out so it can be unit-tested without a live + * connection. The first term avoids underflow in the subtraction. + * + * @param current Subscriptions already tracked on the connection. + * @param additional Subscriptions a request would add. + * @param cap The effective per-connection cap. Defaults to the + * built-in limit; callers may pass a configured override. + * @return true if the request must be rejected to stay within the cap. + */ +[[nodiscard]] constexpr bool +exceedsSubscriptionCap( + std::size_t current, + std::size_t additional, + std::size_t cap = kMaxSubscriptionsPerConnection) +{ + return additional > cap || current > cap - additional; +} + class InfoSubRequest : public CountedObject { public: @@ -44,12 +78,12 @@ public: * map. * * @note Lifetime contract: every `InfoSub` instance MUST be destroyed - * before the backing `Source`. NetworkOPsImp shutdown drops all - * subscriber strong refs before its own teardown to satisfy this. + * before the backing `Source`. NetworkOPsImp shutdown drops all + * subscriber strong refs before its own teardown to satisfy this. * @note Thread-safety: per-instance state is guarded by `lock_`. The - * destructor reads tracking sets without taking `lock_` because - * the strong-pointer ref-count is zero at destruction time, so - * no other thread can be calling the public mutators. + * destructor reads tracking sets without taking `lock_` because + * the strong-pointer ref-count is zero at destruction time, so + * no other thread can be calling the public mutators. */ class InfoSub : public CountedObject { @@ -117,6 +151,34 @@ public: AccountID const& account, bool historyOnly) = 0; + /** + * Schedule the server-side teardown of a disconnecting connection's + * account subscriptions off the destructor thread. + * + * The implementation posts a low-priority JobQueue task that erases the + * entries in bounded chunks, so `~InfoSub` returns immediately instead + * of running the erase loop inline. The sets are taken by value so the + * job owns its copies and never references the destroyed `InfoSub`. + * Cleanup is keyed on `seq` (unique per connection), so deferring it + * cannot disturb a reconnected client reusing the same accounts. + * + * @param seq The disconnecting connection's unique subscription id. + * @param rtAccounts Real-time account subscriptions to remove. + * @param normalAccounts Normal account subscriptions to remove. + * @param historyAccounts Account-history subscriptions to remove. + * + * @note The implementing `Source` must outlive any job it posts. If the + * JobQueue is already stopping (process shutdown), the job is not + * enqueued; the cleanup is skipped because the server-side maps + * are about to be destroyed and no publishing can run. + */ + virtual void + scheduleAccountCleanup( + std::uint64_t seq, + hash_set rtAccounts, + hash_set normalAccounts, + hash_set historyAccounts) = 0; + // VFALCO TODO Document the bool return value virtual bool subLedger(ref ispListener, json::Value& jvResult) = 0; @@ -153,12 +215,12 @@ public: * @param ispListener The subscriber requesting removal. * @param book The order book to unsubscribe from. * @return true if the entry was present and removed, false if the - * subscriber was not subscribed to @p book. + * subscriber was not subscribed to @p book. * - * @note Thread-safety: acquires subLock_ internally. + * @note Thread-safety: acquires bookLock_ internally. * @note Do NOT call from ~InfoSub(). Use unsubBookInternal instead - * to avoid a redundant write-back to bookSubscriptions_ on a - * partially-destroyed object. + * to avoid a redundant write-back to bookSubscriptions_ on a + * partially-destroyed object. */ virtual bool unsubBook(ref ispListener, Book const&) = 0; @@ -173,9 +235,9 @@ public: * @param uListener The sequence number of the subscriber being torn down. * @param book The order book entry to remove. * @return true if the entry was present and removed, false otherwise - * (e.g., already removed by a concurrent RPC unsubscribe). + * (e.g., already removed by a concurrent RPC unsubscribe). * - * @note Thread-safety: acquires subLock_ internally. + * @note Thread-safety: acquires bookLock_ internally. */ virtual bool unsubBookInternal(std::uint64_t uListener, Book const&) = 0; @@ -221,8 +283,8 @@ public: /** * Journal used by InfoSub for diagnostics that occur after the - * owning subsystem (e.g. application-level Logs) is the only - * surviving sink — primarily destructor-time cleanup failures. + * owning subsystem (e.g. application-level Logs) is the only + * surviving sink — primarily destructor-time cleanup failures. */ [[nodiscard]] virtual beast::Journal const& journal() const = 0; @@ -243,6 +305,56 @@ public: [[nodiscard]] std::uint64_t getSeq() const; + /** + * Return the number of subscriptions currently tracked on this + * connection. + * + * The combined size of the per-connection account, real-time account, and + * account-history subscription sets. `doSubscribe` reads this to enforce + * the per-connection subscription cap before admitting more. + * + * @return The total tracked subscription count for this connection. + * + * @note Thread-safe: takes `lock_` for the read; read-only. + */ + [[nodiscard]] std::size_t + totalSubscriptionCount() const; + + /** + * Enforce the cap and reserve a request's net-new accounts, atomically. + * + * Under one hold of `lock_`: count the net-new entries in the two sets, + * check the total against @p cap, and insert them only if it fits. + * All-or-nothing. Doing check and insert together stops two concurrent + * requests sharing an InfoSub (the admin subscribe-by-url path) from both + * passing the check before either records its accounts. The server-side + * maps are populated afterwards by subAccount, whose re-insert is a no-op. + * + * @param proposedAccounts Real-time (accounts_proposed) ids to reserve. + * @param normalAccounts Normal (accounts) ids to reserve. + * @param cap The effective per-connection cap. + * @return true if reserved; false if the request must be rejected. + * @note Thread-safe: takes `lock_`. + */ + [[nodiscard]] bool + tryReserveAccountSubscriptions( + hash_set const& proposedAccounts, + hash_set const& normalAccounts, + std::size_t cap); + + /** + * Whether this connection already tracks an account-history for @p account. + * + * `doSubscribe` reads this to charge the cap for an account_history_tx_stream + * only when it is net-new, matching the account branches. + * + * @param account The account an account_history_tx_stream would add. + * @return true if @p account is already in the account-history set. + * @note Thread-safe: takes `lock_`; read-only. + */ + [[nodiscard]] bool + hasAccountHistorySubscription(AccountID const& account) const; + void onSendEmpty(); @@ -302,7 +414,9 @@ public: getApiVersion() const noexcept; protected: - std::mutex lock_; + // Mutable so the read-only totalSubscriptionCount() accessor can lock it + // from a const method; locking semantics are otherwise unchanged. + mutable std::mutex lock_; private: Consumer consumer_; diff --git a/src/libxrpl/server/InfoSub.cpp b/src/libxrpl/server/InfoSub.cpp index 353c295856..ceb1027d95 100644 --- a/src/libxrpl/server/InfoSub.cpp +++ b/src/libxrpl/server/InfoSub.cpp @@ -7,10 +7,12 @@ #include #include +#include #include #include #include #include +#include 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 const& proposedAccounts, + hash_set 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 const& requested, + hash_set 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) { diff --git a/src/test/rpc/Subscribe_test.cpp b/src/test/rpc/Subscribe_test.cpp index 97c5290947..567f31437a 100644 --- a/src/test/rpc/Subscribe_test.cpp +++ b/src/test/rpc/Subscribe_test.cpp @@ -27,6 +27,7 @@ #include #include #include +#include #include #include #include @@ -1548,6 +1549,413 @@ public: } } + // ----- Subscription limit / teardown verification ---------------------- + // + // The helpers and tests below exercise: + // * the per-connection subscription cap + proportional charge enforced + // in doSubscribe (Subscribe.cpp), and + // * the asynchronous, chunked teardown of a disconnecting connection's + // account subscriptions (~InfoSub -> scheduleAccountCleanup -> JobQueue). + // + // The cap-exceeded error is rpcINVALID_PARAMS with the message "Too many + // subscriptions for this connection."; the tests assert that exactly. + // + // There is no public accessor for the server-side per-connection count, so + // the async cleanup is verified behaviorally: publishing still flows to a + // live subscriber, rather than by reading a count to zero. + + // Build `count` distinct, valid, base58-encoded account strings cheaply by + // incrementing an AccountID. parseAccountIds dedups into a hash_set, so the + // strings MUST be distinct for the cap arithmetic to be exact; incrementing + // guarantees distinctness without deriving `count` keypairs. + static std::vector + makeAccountStrings(std::size_t count, std::uint32_t seed = 1) + { + std::vector out; + out.reserve(count); + // Start at `seed` so separate calls produce non-overlapping ranges, + // letting a test subscribe disjoint batches across requests. + AccountID id{static_cast(seed)}; + for (std::size_t i = 0; i < count; ++i) + { + out.push_back(toBase58(id)); + ++id; + } + return out; + } + + // Append the given account strings as a jss::accounts array onto a fresh + // subscribe request object. + static json::Value + accountsRequest(std::vector const& accts) + { + json::Value jv{json::ValueType::Object}; + jv[jss::accounts] = json::ValueType::Array; + for (auto const& a : accts) + jv[jss::accounts].append(a); + return jv; + } + + // Append the given account strings as a jss::accounts_proposed array onto a + // fresh subscribe request object. + static json::Value + accountsProposedRequest(std::vector const& accts) + { + json::Value jv{json::ValueType::Object}; + jv[jss::accounts_proposed] = json::ValueType::Array; + for (auto const& a : accts) + jv[jss::accounts_proposed].append(a); + return jv; + } + + // A single, valid XRP/USD order book request, as one entry of a + // jss::books array. + static json::Value + oneBookRequest() + { + using namespace jtx; + json::Value jv{json::ValueType::Object}; + jv[jss::books] = json::ValueType::Array; + json::Value& book = jv[jss::books][0u]; + book[jss::taker_gets] = json::ValueType::Object; + book[jss::taker_gets][jss::currency] = "XRP"; + book[jss::taker_pays] = json::ValueType::Object; + book[jss::taker_pays][jss::currency] = "USD"; + book[jss::taker_pays][jss::issuer] = Account("alice").human(); + return jv; + } + + // A single account_history_tx_stream subscribe request for `acct`. + static json::Value + accountHistoryRequest(std::string const& acct) + { + json::Value jv{json::ValueType::Object}; + jv[jss::account_history_tx_stream] = json::ValueType::Object; + jv[jss::account_history_tx_stream][jss::account] = acct; + return jv; + } + + // An envconfig modifier that lowers the per-connection subscription cap to + // `cap`, so the cap logic in doSubscribe can be driven without subscribing + // the production default (100'000) entries. (Env is non-movable, so this + // returns the config modifier rather than a ready-made Env.) + static auto + cappedConfig(std::size_t cap) + { + return [cap](std::unique_ptr cfg) { + cfg->maxSubscriptionsPerConnection = cap; + return jtx::singleThreadIo(std::move(cfg)); + }; + } + + void + testSubscriptionCapRejects() + { + // A request that alone exceeds the cap is rejected with the exact + // cap error, before any state is recorded. Baseline negative path. + testcase("subscription cap rejects an over-cap request"); + + using namespace jtx; + Env env{*this, envconfig(cappedConfig(5))}; + auto wsc = makeWSClient(env.app().config()); + + // Six accounts against a cap of five: rejected. + auto const jr = + wsc->invoke("subscribe", accountsRequest(makeAccountStrings(6)))[jss::result]; + BEAST_EXPECT(jr[jss::error] == "invalidParams"); + BEAST_EXPECT(jr[jss::error_message] == "Too many subscriptions for this connection."); + } + + void + testReSubscribeNotOvercounted() + { + // Re-subscribing accounts already held by this connection adds no new + // tracked state, so it must be admitted even at the cap. The cap check + // must count only NET-NEW accounts, not the raw request size. + testcase("re-subscribe at the cap is not over-counted"); + + using namespace jtx; + Env env{*this, envconfig(cappedConfig(5))}; + auto wsc = makeWSClient(env.app().config()); + + // Fill the cap exactly with five distinct accounts. + auto const five = makeAccountStrings(5); + { + auto const r = wsc->invoke("subscribe", accountsRequest(five)); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + + // Re-subscribe the same five: net-new is zero, so it stays within the + // cap and must succeed. (Pre-fix this was wrongly rejected.) + { + auto const r = wsc->invoke("subscribe", accountsRequest(five)); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + } + + void + testBooksCapIndependentOfAccounts() + { + // Book subscriptions are tracked separately (OrderBookDB) and are not + // part of totalSubscriptionCount(). An account set at the cap must not + // block an unrelated book subscription. + testcase("books cap is independent of account count"); + + using namespace jtx; + Env env{*this, envconfig(cappedConfig(5))}; + Account const alice{"alice"}; + env.fund(XRP(10000), alice); + BEAST_EXPECT(env.syncClose()); + + auto wsc = makeWSClient(env.app().config()); + + // Fill the account cap exactly. + { + auto const r = wsc->invoke("subscribe", accountsRequest(makeAccountStrings(5))); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + + // A single book subscription must still be admitted: it does not count + // against the account cap. (Pre-fix this was wrongly rejected.) + { + auto const r = wsc->invoke("subscribe", oneBookRequest()); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + } + + void + testMultiFieldNoPartialSubscribe() + { + // A single request mixing fields must be all-or-nothing: if a later + // field trips the cap, an earlier field must NOT have subscribed. The + // leak is detected through the cap arithmetic itself - a follow-up + // request succeeds only if no state leaked from the rejected one. + testcase("multi-field subscribe does not partially subscribe"); + + using namespace jtx; + Env env{*this, envconfig(cappedConfig(5))}; + auto wsc = makeWSClient(env.app().config()); + + // accounts_proposed (3, evaluated first, would subscribe) + + // accounts (3): combined 6 exceeds the cap of 5, so the request is + // rejected. The proposed branch must not have leaked its 3 entries. + json::Value req = accountsProposedRequest(makeAccountStrings(3, 1)); + for (auto const& a : makeAccountStrings(3, 100)) + req[jss::accounts].append(a); + { + auto const jr = wsc->invoke("subscribe", req)[jss::result]; + BEAST_EXPECT(jr[jss::error] == "invalidParams"); + BEAST_EXPECT(jr[jss::error_message] == "Too many subscriptions for this connection."); + } + + // If the rejected request leaked its 3 proposed subscriptions, the + // connection's count is already 3 and this 3-account request would be + // rejected (3 + 3 > 5). With no leak the count is 0 and it succeeds. + { + auto const r = wsc->invoke("subscribe", accountsRequest(makeAccountStrings(3, 200))); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + } + + void + testHistoryReSubscribeNotOvercounted() + { + // An account_history_tx_stream subscribe is charged against the cap only + // when it is net-new, matching the account branches. Re-subscribing an + // account-history already held on this connection adds no tracked entry, + // so it must NOT be rejected at the cap. The two rejection causes are + // told apart by their exact error_message: the cap check yields "Too + // many subscriptions for this connection."; a duplicate that gets past + // the cap and is rejected downstream by subAccountHistory yields the + // generic "Invalid parameters.". + testcase("account_history re-subscribe at the cap is not over-counted"); + + using namespace jtx; + Env env{*this, envconfig(cappedConfig(1))}; + Account const alice{"alice"}; + env.fund(XRP(10000), alice); + BEAST_EXPECT(env.syncClose()); + + auto wsc = makeWSClient(env.app().config()); + + // First account-history subscribe is net-new: charge 1 fills the cap of + // 1 exactly, so it is admitted. Positive path. + { + auto const r = wsc->invoke("subscribe", accountHistoryRequest(alice.human())); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + + // Re-subscribe the same account-history while sitting exactly at the + // cap. Net-new is zero, so the cap check must pass; the request is then + // rejected by subAccountHistory as a duplicate, NOT by the cap. Proven + // by the exact message: it is the duplicate error, not the cap error. + // (Pre-fix, the flat charge of 1 made the cap check reject this with the + // cap message instead.) + { + auto const jr = + wsc->invoke("subscribe", accountHistoryRequest(alice.human()))[jss::result]; + BEAST_EXPECT(jr[jss::error] == "invalidParams"); + BEAST_EXPECT(jr[jss::error_message] == "Invalid parameters."); + BEAST_EXPECT(jr[jss::error_message] != "Too many subscriptions for this connection."); + } + } + + void + testHistoryCapRejectsNetNew() + { + // A genuinely net-new account-history subscribe on a connection already + // at the cap IS rejected, with the cap error. Negative path, and the + // counterpart to testHistoryReSubscribeNotOvercounted: it confirms the + // net-new charge still rejects when the entry really is new. + testcase("account_history net-new subscribe is rejected at the cap"); + + using namespace jtx; + Env env{*this, envconfig(cappedConfig(1))}; + Account const alice{"alice"}; + Account const bob{"bob"}; + env.fund(XRP(10000), alice, bob); + BEAST_EXPECT(env.syncClose()); + + auto wsc = makeWSClient(env.app().config()); + + // Fill the cap of 1 with alice's account-history. + { + auto const r = wsc->invoke("subscribe", accountHistoryRequest(alice.human())); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + + // A different account-history (bob) is net-new: charge 1 over a cap of 1 + // already full, so it is rejected with the cap error. + { + auto const jr = + wsc->invoke("subscribe", accountHistoryRequest(bob.human()))[jss::result]; + BEAST_EXPECT(jr[jss::error] == "invalidParams"); + BEAST_EXPECT(jr[jss::error_message] == "Too many subscriptions for this connection."); + } + } + + void + testAsyncTeardownDoesNotStall() + { + // Test C (core regression): disconnecting a connection with many + // account subscriptions must NOT block subsequent operations or + // publishing. The teardown is now posted to a JobQueue job + // (scheduleAccountCleanup), so it runs off the disconnect thread. + testcase("async teardown does not stall publishing"); + + using namespace std::chrono_literals; + using namespace jtx; + Env env{*this, singleThreadIo(envconfig())}; + + Account const alice{"alice"}; + env.fund(XRP(10000), alice); + BEAST_EXPECT(env.syncClose()); + + // A second, long-lived subscriber to alice that must keep receiving + // publishes after the first connection disconnects. + auto wscLive = makeWSClient(env.app().config()); + { + json::Value jv{json::ValueType::Object}; + jv[jss::accounts] = json::ValueType::Array; + jv[jss::accounts].append(alice.human()); + auto const r = wscLive->invoke("subscribe", jv); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + + // A connection that subscribes to many accounts, then disconnects. A + // few thousand entries is enough to be a real teardown while still + // running fast in CI. + constexpr std::size_t kBulk = 3000; + { + auto wscBulk = makeWSClient(env.app().config()); + auto const r = + wscBulk->invoke("subscribe", accountsRequest(makeAccountStrings(kBulk, 10))); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + // Destroying the client closes the WS connection, which destroys + // the server-side InfoSub and posts the chunked async cleanup job. + // WSClient exposes no explicit close(); resetting the owning + // unique_ptr is the disconnect path. + wscBulk.reset(); + } + + // Immediately after the disconnect, an unrelated operation completes + // promptly (it would block for seconds with inline teardown). This is a + // cheap liveness check; the publish assertion below is the real proof. + { + auto const info = env.app().getOPs().getServerInfo(false, true, false); + BEAST_EXPECT(info.isMember(jss::server_state)); + } + + // The live subscriber still receives a published transaction for alice + // within a short timeout, proving account-publishing was not stalled by + // the concurrent teardown. + { + env(pay(env.master, alice, XRP(100))); + BEAST_EXPECT(env.syncClose()); + BEAST_EXPECT(wscLive->findMsg(5s, [&](auto const& jv) { + return jv.isMember(jss::transaction) && + jv[jss::transaction][jss::TransactionType] == jss::Payment && + jv[jss::transaction][jss::Destination] == alice.human(); + })); + } + + wscLive->invoke("unsubscribe", accountsRequest({alice.human()})); + } + + void + testResubscribeAfterDisconnect() + { + // Test D (Phase 3 correctness): connection A subscribes to account X + // and disconnects (async cleanup pending, keyed on A's seq). A new + // connection B subscribes to X and MUST still receive publishes for X - + // A's deferred, seq-keyed cleanup must not remove B's subscription. + testcase("re-subscribe after disconnect still delivers"); + + using namespace std::chrono_literals; + using namespace jtx; + Env env{*this, singleThreadIo(envconfig())}; + + Account const alice{"alice"}; + env.fund(XRP(10000), alice); + BEAST_EXPECT(env.syncClose()); + + // Connection A subscribes to alice, then disconnects. A also subscribes + // to a bulk set so its deferred cleanup is non-trivial and races with B. + { + auto wscA = makeWSClient(env.app().config()); + auto bulk = makeAccountStrings(2000, 10); + bulk.push_back(alice.human()); + auto const r = wscA->invoke("subscribe", accountsRequest(bulk)); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + // Disconnect A by destroying its client (no explicit close()). + wscA.reset(); + } + + // Connection B (a new InfoSub with a distinct seq) subscribes to alice. + auto wscB = makeWSClient(env.app().config()); + { + json::Value jv{json::ValueType::Object}; + jv[jss::accounts] = json::ValueType::Array; + jv[jss::accounts].append(alice.human()); + auto const r = wscB->invoke("subscribe", jv); + BEAST_EXPECTS(r[jss::status] == "success", to_string(r)); + } + + // A publish for alice must reach B. If A's seq-keyed cleanup had wrongly + // removed the shared alice entry, B would receive nothing. + { + env(pay(env.master, alice, XRP(100))); + BEAST_EXPECT(env.syncClose()); + BEAST_EXPECT(wscB->findMsg(5s, [&](auto const& jv) { + return jv.isMember(jss::transaction) && + jv[jss::transaction][jss::TransactionType] == jss::Payment && + jv[jss::transaction][jss::Destination] == alice.human(); + })); + } + + wscB->invoke("unsubscribe", accountsRequest({alice.human()})); + } + void run() override { @@ -1569,6 +1977,14 @@ public: testSubBookChanges(); testNFToken(all); testNFToken(all - featureNFTokenMintOffer); + testAsyncTeardownDoesNotStall(); + testResubscribeAfterDisconnect(); + testSubscriptionCapRejects(); + testReSubscribeNotOvercounted(); + testBooksCapIndependentOfAccounts(); + testMultiFieldNoPartialSubscribe(); + testHistoryReSubscribeNotOvercounted(); + testHistoryCapRejectsNetNew(); } }; diff --git a/src/tests/libxrpl/CMakeLists.txt b/src/tests/libxrpl/CMakeLists.txt index cafe72eff9..2fe046f3d4 100644 --- a/src/tests/libxrpl/CMakeLists.txt +++ b/src/tests/libxrpl/CMakeLists.txt @@ -33,6 +33,7 @@ set(test_modules shamap tx protocol_autogen + server ) if(NOT WIN32) list(APPEND test_modules net) diff --git a/src/tests/libxrpl/server/InfoSub.cpp b/src/tests/libxrpl/server/InfoSub.cpp new file mode 100644 index 0000000000..6913812a92 --- /dev/null +++ b/src/tests/libxrpl/server/InfoSub.cpp @@ -0,0 +1,60 @@ +#include + +#include + +#include +#include + +using namespace xrpl; + +// The per-connection subscription cap is enforced by the pure predicate +// exceedsSubscriptionCap(current, additional). Testing it directly (rather than +// by subscribing the real cap through a WebSocket, which would exceed the frame +// limit and drop the connection before the check runs) lets the boundary be +// asserted exactly. +TEST(InfoSubSubscriptionCap, Boundary) +{ + constexpr std::size_t cap = kMaxSubscriptionsPerConnection; + + // Empty connection: anything up to the cap is admitted, cap+1 is not. + EXPECT_FALSE(exceedsSubscriptionCap(0, 0)); + EXPECT_FALSE(exceedsSubscriptionCap(0, cap)); + EXPECT_TRUE(exceedsSubscriptionCap(0, cap + 1)); + + // Exactly at the cap: zero more is fine, one more is rejected. + EXPECT_FALSE(exceedsSubscriptionCap(cap, 0)); + EXPECT_TRUE(exceedsSubscriptionCap(cap, 1)); + + // One below the cap: exactly one more reaches the cap; two exceed it. + EXPECT_FALSE(exceedsSubscriptionCap(cap - 1, 1)); + EXPECT_TRUE(exceedsSubscriptionCap(cap - 1, 2)); +} + +TEST(InfoSubSubscriptionCap, NoOverflow) +{ + constexpr std::size_t cap = kMaxSubscriptionsPerConnection; + constexpr std::size_t max = std::numeric_limits::max(); + + // current + additional must not wrap: a huge additional is rejected even + // when current is 0 (the additional > cap term guards the subtraction). + EXPECT_TRUE(exceedsSubscriptionCap(0, max)); + EXPECT_TRUE(exceedsSubscriptionCap(cap, max)); +} + +TEST(InfoSubSubscriptionCap, ExplicitCap) +{ + // A configured override is honored: the boundary tracks the passed cap, not + // the built-in default. This is the seam doSubscribe uses to enforce a + // per-connection cap set via [max_subscriptions_per_connection]. + constexpr std::size_t cap = 5; + + EXPECT_FALSE(exceedsSubscriptionCap(0, cap, cap)); + EXPECT_TRUE(exceedsSubscriptionCap(0, cap + 1, cap)); + EXPECT_FALSE(exceedsSubscriptionCap(cap, 0, cap)); + EXPECT_TRUE(exceedsSubscriptionCap(cap, 1, cap)); + EXPECT_FALSE(exceedsSubscriptionCap(cap - 1, 1, cap)); + EXPECT_TRUE(exceedsSubscriptionCap(cap - 1, 2, cap)); + + // The overflow guard still holds with a small explicit cap. + EXPECT_TRUE(exceedsSubscriptionCap(0, std::numeric_limits::max(), cap)); +} diff --git a/src/xrpld/app/ledger/AcceptedLedger.h b/src/xrpld/app/ledger/AcceptedLedger.h index 6e42d611d4..a8b78d08b0 100644 --- a/src/xrpld/app/ledger/AcceptedLedger.h +++ b/src/xrpld/app/ledger/AcceptedLedger.h @@ -57,6 +57,15 @@ public: return transactions_.end(); } + /** + * The last accepted transaction. Precondition: size() > 0. + */ + [[nodiscard]] AcceptedLedgerTx const& + back() const + { + return *transactions_.back(); + } + private: std::shared_ptr ledger_; std::vector> transactions_; diff --git a/src/xrpld/app/misc/NetworkOPs.cpp b/src/xrpld/app/misc/NetworkOPs.cpp index 4b0091dff6..b43149b5af 100644 --- a/src/xrpld/app/misc/NetworkOPs.cpp +++ b/src/xrpld/app/misc/NetworkOPs.cpp @@ -154,6 +154,12 @@ namespace xrpl { +/** + * Concrete NetworkOPs: server sequencer, network tracker, and owner of all + * client subscription state (accounts, books, streams). Subscriptions use three + * independent non-recursive locks (accountLock_, bookLock_, streamLock_); see + * their declarations for the locking and deferred-destruction rules. + */ class NetworkOPsImp final : public NetworkOPs { /** @@ -194,7 +200,7 @@ class NetworkOPsImp final : public NetworkOPs /** * State accounting records two attributes for each possible server state: * 1) Amount of time spent in each state (in microseconds). This value is - * updated upon each state transition. + * updated upon each state transition. * 2) Number of transitions to each state. * * This data can be polled through server_info and represented by @@ -573,6 +579,13 @@ public: unsubAccountHistoryInternal(std::uint64_t seq, AccountID const& account, bool historyOnly) override; + void + scheduleAccountCleanup( + std::uint64_t seq, + hash_set rtAccounts, + hash_set normalAccounts, + hash_set historyAccounts) override; + bool subLedger(InfoSub::ref ispListener, json::Value& jvResult) override; bool @@ -636,6 +649,20 @@ public: bool tryRemoveRpcSub(std::string const& strUrl) override; + /** + * Look up an RPC subscription without taking streamLock_. + * + * Callers MUST already hold streamLock_. This exists so tryRemoveRpcSub + * can reuse the lookup while holding the lock; the plain std::mutex is not + * recursive, so calling the public findRpcSub (which locks) from under the + * lock would self-deadlock. + * + * @param strUrl The subscription URL key into rpcSubMap_. + * @return The matching InfoSub, or an empty pointer if not found. + */ + InfoSub::pointer + findRpcSubLocked(std::string const& strUrl); + beast::Journal const& journal() const override { @@ -724,22 +751,22 @@ private: * Extracts the set of order books affected by @p transaction, then * delivers @p jvObj to every live subscriber of those books. * - * Uses a two-pass design to keep subLock_ hold time short: - * 1. Under subLock_, collect strong InfoSub pointers for all live - * subscribers and prune any expired weak_ptrs encountered. - * 2. Release subLock_, then call send() on each collected pointer. + * Uses a two-pass design to keep bookLock_ hold time short: + * 1. Under bookLock_, collect strong InfoSub pointers for all live + * subscribers and prune any expired weak_ptrs encountered. + * 2. Release bookLock_, then call send() on each collected pointer. * * @param transaction The accepted ledger transaction to inspect. * @param jvObj JSON representation of the transaction to deliver. * - * @note Thread-safety: acquires subLock_ for the collection pass only. - * send() is intentionally called outside the lock to avoid blocking - * all other sub/unsub/publish paths while I/O is in progress. - * @note Contention: subLock_ is shared with all other subscription types. - * On high-throughput nodes processing multi-hop payments that touch - * many offer nodes, this pass holds subLock_ longer than the old - * per-book BookListeners locks did. This is an accepted trade-off - * for lock-domain simplicity. + * @note Thread-safety: acquires bookLock_ for the collection pass only. + * send() is intentionally called outside the lock to avoid blocking + * other book sub/unsub/publish paths while I/O is in progress. + * @note Contention: bookLock_ guards only book subscriptions, so this pass + * no longer competes with account or stream traffic. On high-throughput + * nodes processing multi-hop payments that touch many offer nodes, it + * still holds bookLock_ longer than the old per-book BookListeners + * locks did. This is an accepted trade-off for lock-domain simplicity. */ void pubBookTransaction(AcceptedLedgerTx const& transaction, MultiApiJson const& jvObj); @@ -750,6 +777,23 @@ private: std::shared_ptr const& transaction, TER result); + /** + * Send the ledgerClosed and book-changes stream updates for a ledger. + * Takes streamLock_ only. + */ + void + publishLedgerStreams( + std::shared_ptr const& lpAccepted, + std::shared_ptr const& alpAccepted); + + /** + * On the first published ledger only, start the delayed account-history + * streaming for any subscriptions that were registered before a validated + * ledger existed. Takes accountLock_ only. + */ + void + kickoffAccountHistory(std::shared_ptr const& alpAccepted); + void pubServer(); void @@ -802,7 +846,9 @@ private: hash_map>; /** - * @note called while holding subLock_ + * @note called while holding accountLock_ (it only touches + * subAccountHistory_ and posts a JobQueue task; it never reacquires + * a subscription lock nor touches the stream maps). */ void subAccountHistoryStart( @@ -813,12 +859,85 @@ private: void setAccountHistoryJobTimer(SubAccountHistoryInfoWeak subInfo); + /** + * Maximum number of account entries erased per accountLock_ acquisition + * during disconnect-time cleanup. + * + * The cleanup erase loops drop and reacquire accountLock_ after every + * chunk of this many accounts, bounding how long a large teardown holds + * the lock. A concurrent publish may interleave between chunks; that is + * safe because publishing tolerates a partially-cleaned map (a dead + * subscriber is simply not notified). + */ + static constexpr std::size_t kAccountCleanupChunk = 4096; + + /** + * Erase one connection's entries from a subscription map in + * accountLock_-bounded chunks. + * + * Shared engine behind cleanupAccountSubscriptions and + * cleanupAccountHistorySubscriptions: both walk @p accounts, and for each + * remove this connection's @p seq from the inner per-account map, dropping + * the outer entry once its last subscriber leaves. The lock is released + * between chunks so a competing publish can interleave; no iterator is held + * across the unlock, so a concurrent mutation cannot dangle. + * + * @tparam OuterMap hash_map>. + * @tparam BeforeErase Invoked with the inner value about to be erased, for + * per-entry teardown the plain account maps do not need + * (the history map uses it to stop its paging job). + * @param seq The disconnecting connection's subscription id. + * @param accounts The accounts this connection was subscribed to. + * @param outerMap The subscription map to erase from. + * @param beforeErase Called on each inner value just before it is erased. + * See kAccountCleanupChunk. + */ + template + void + cleanupSubscriptionMap( + std::uint64_t seq, + hash_set const& accounts, + OuterMap& outerMap, + BeforeErase&& beforeErase); + + /** + * Erase one connection's entries from the given account map (subAccount_ + * or subRTAccount_) in accountLock_-bounded chunks. The caller selects the + * map, so this need not know about the real-time/normal distinction. Keyed + * on seq, so it only removes the disconnecting connection's entries. + * See kAccountCleanupChunk. + */ + void + cleanupAccountSubscriptions( + std::uint64_t seq, + hash_set const& accounts, + SubInfoMapType& subMap); + + /** + * Erase one connection's entries from subAccountHistory_ in + * accountLock_-bounded chunks. Keyed on seq. See kAccountCleanupChunk. + */ + void + cleanupAccountHistorySubscriptions(std::uint64_t seq, hash_set const& accounts); + std::reference_wrapper registry_; beast::Journal journal_; std::unique_ptr localTX_; - std::recursive_mutex subLock_; + // Independent lock domains so a long cleanup/publish on one does not stall + // the others. Hold at most one at a time; if ever more, order: accountLock_, + // bookLock_, streamLock_. + // + // Deferred-destruction rule (non-recursive mutexes): under bookLock_ or + // streamLock_, never let the last InfoSub pointer die inside the lock - + // ~InfoSub re-acquires it via unsub* -> self-deadlock. Publishers collect the + // locked pointers in a vector declared before the lock and destruct after + // release (see pubServer / pubBookTransaction). accountLock_ is exempt: + // ~InfoSub offloads account teardown to scheduleAccountCleanup. + std::mutex accountLock_; ///< Guards subAccount_, subRTAccount_, subAccountHistory_. + std::mutex bookLock_; ///< Guards subBook_. + std::mutex streamLock_; ///< Guards streamMaps_[] and rpcSubMap_. std::atomic mode_; @@ -843,18 +962,18 @@ private: /** * Maps each order book to its current set of subscribers. - * Outer key: the Book (currency pair + optional domain). - * Inner key: InfoSub::seq (unique per connection). - * Inner value: weak_ptr so that a dropped connection does not prevent - * the InfoSub from being destroyed; expired entries are pruned lazily - * by pubBookTransaction and eagerly by unsubBookInternal (~InfoSub path). - * Guarded by subLock_. + * Outer key: the Book (currency pair + optional domain). + * Inner key: InfoSub::seq (unique per connection). + * Inner value: weak_ptr so that a dropped connection does not prevent + * the InfoSub from being destroyed; expired entries are pruned lazily + * by pubBookTransaction and eagerly by unsubBookInternal (~InfoSub path). + * Guarded by bookLock_. */ using SubBookMapType = hash_map; SubInfoMapType subAccount_; SubInfoMapType subRTAccount_; - SubBookMapType subBook_; ///< Guarded by subLock_. + SubBookMapType subBook_; ///< Guarded by bookLock_. subRpcMapType rpcSubMap_; @@ -875,6 +994,10 @@ private: SLastEntry // Any new entry must be ADDED ABOVE this one }; + /** + * One weak_ptr subscriber map per stream type. Guarded by streamLock_; + * subject to its deferred-destruction rule (see pubServer). + */ std::array streamMaps_; ServerFeeSummary lastFeeSummary_; @@ -2245,8 +2368,14 @@ NetworkOPsImp::consensusViewChange() void NetworkOPsImp::pubManifest(Manifest const& mo) { + // Hold each locked subscriber alive until after streamLock_ is released: + // if this is the last reference, ~InfoSub re-acquires streamLock_ (via its + // unsub* calls), which would self-deadlock on this non-recursive mutex. + // Declared before the lock so it is destroyed after the lock is dropped. + std::vector toRelease; + // VFALCO consider std::shared_mutex - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); if (!streamMaps_[SManifests].empty()) { @@ -2269,6 +2398,7 @@ NetworkOPsImp::pubManifest(Manifest const& mo) if (auto p = i->second.lock()) { p->send(jvObj, true); + toRelease.push_back(std::move(p)); ++i; } else @@ -2320,11 +2450,16 @@ trunc32(std::uint64_t v) void NetworkOPsImp::pubServer() { + // Hold each locked subscriber alive until after streamLock_ is released; a + // last-reference ~InfoSub would otherwise re-acquire this non-recursive + // mutex and self-deadlock. Declared before the lock, destroyed after it. + std::vector toRelease; + // VFALCO TODO Don't hold the lock across calls to send...make a copy of the // list into a local array while holding the lock then release // the lock and call send on everyone. // - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); if (!streamMaps_[SServer].empty()) { @@ -2362,7 +2497,7 @@ NetworkOPsImp::pubServer() for (auto i = streamMaps_[SServer].begin(); i != streamMaps_[SServer].end();) { - InfoSub::pointer const p = i->second.lock(); + InfoSub::pointer p = i->second.lock(); // VFALCO TODO research the possibility of using thread queues and // linearizing the deletion of subscribers with the @@ -2370,6 +2505,7 @@ NetworkOPsImp::pubServer() if (p) { p->send(jvObj, true); + toRelease.push_back(std::move(p)); ++i; } else @@ -2383,7 +2519,12 @@ NetworkOPsImp::pubServer() void NetworkOPsImp::pubConsensus(ConsensusPhase phase) { - std::scoped_lock const sl(subLock_); + // Hold each locked subscriber alive until after streamLock_ is released; a + // last-reference ~InfoSub would otherwise re-acquire this non-recursive + // mutex and self-deadlock. Declared before the lock, destroyed after it. + std::vector toRelease; + + std::scoped_lock const sl(streamLock_); auto& streamMap = streamMaps_[SConsensusPhase]; if (!streamMap.empty()) @@ -2397,6 +2538,7 @@ NetworkOPsImp::pubConsensus(ConsensusPhase phase) if (auto p = i->second.lock()) { p->send(jvObj, true); + toRelease.push_back(std::move(p)); ++i; } else @@ -2410,8 +2552,13 @@ NetworkOPsImp::pubConsensus(ConsensusPhase phase) void NetworkOPsImp::pubValidation(std::shared_ptr const& val) { + // Hold each locked subscriber alive until after streamLock_ is released; a + // last-reference ~InfoSub would otherwise re-acquire this non-recursive + // mutex and self-deadlock. Declared before the lock, destroyed after it. + std::vector toRelease; + // VFALCO consider std::shared_mutex - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); if (!streamMaps_[SValidations].empty()) { @@ -2503,6 +2650,7 @@ NetworkOPsImp::pubValidation(std::shared_ptr const& val) multiObj.visit( p->getApiVersion(), // [&](json::Value const& jv) { p->send(jv, true); }); + toRelease.push_back(std::move(p)); ++i; } else @@ -2516,7 +2664,12 @@ NetworkOPsImp::pubValidation(std::shared_ptr const& val) void NetworkOPsImp::pubPeerStatus(std::function const& func) { - std::scoped_lock const sl(subLock_); + // Hold each locked subscriber alive until after streamLock_ is released; a + // last-reference ~InfoSub would otherwise re-acquire this non-recursive + // mutex and self-deadlock. Declared before the lock, destroyed after it. + std::vector toRelease; + + std::scoped_lock const sl(streamLock_); if (!streamMaps_[SPeerStatus].empty()) { @@ -2526,11 +2679,12 @@ NetworkOPsImp::pubPeerStatus(std::function const& func) for (auto i = streamMaps_[SPeerStatus].begin(); i != streamMaps_[SPeerStatus].end();) { - InfoSub::pointer const p = i->second.lock(); + InfoSub::pointer p = i->second.lock(); if (p) { p->send(jvObj, true); + toRelease.push_back(std::move(p)); ++i; } else @@ -3074,7 +3228,13 @@ NetworkOPsImp::pubProposedTransaction( MultiApiJson const jvObj = transJson(transaction, result, false, ledger, std::nullopt); { - std::scoped_lock const sl(subLock_); + // Hold each locked subscriber alive until after streamLock_ is + // released; a last-reference ~InfoSub would otherwise re-acquire this + // non-recursive mutex and self-deadlock. Declared before the lock, + // destroyed after the block ends. + std::vector toRelease; + + std::scoped_lock const sl(streamLock_); auto it = streamMaps_[SRtTransactions].begin(); while (it != streamMaps_[SRtTransactions].end()) @@ -3086,6 +3246,7 @@ NetworkOPsImp::pubProposedTransaction( jvObj.visit( p->getApiVersion(), // [&](json::Value const& jv) { p->send(jv, true); }); + toRelease.push_back(std::move(p)); ++it; } else @@ -3117,100 +3278,121 @@ NetworkOPsImp::pubLedger(std::shared_ptr const& lpAccepted) alpAccepted->getLedger().get() == lpAccepted.get(), "xrpl::NetworkOPsImp::pubLedger : accepted input"); - { - JLOG(journal_.debug()) << "Publishing ledger " << lpAccepted->header().seq << " " - << lpAccepted->header().hash; + JLOG(journal_.debug()) << "Publishing ledger " << lpAccepted->header().seq << " " + << lpAccepted->header().hash; - std::scoped_lock const sl(subLock_); - - if (!streamMaps_[SLedger].empty()) - { - json::Value jvObj(json::ValueType::Object); - - jvObj[jss::type] = "ledgerClosed"; - jvObj[jss::ledger_index] = lpAccepted->header().seq; - jvObj[jss::ledger_hash] = to_string(lpAccepted->header().hash); - jvObj[jss::ledger_time] = - json::Value::UInt(lpAccepted->header().closeTime.time_since_epoch().count()); - - jvObj[jss::network_id] = registry_.get().getNetworkIDService().getNetworkID(); - - if (!lpAccepted->rules().enabled(featureXRPFees)) - jvObj[jss::fee_ref] = kFeeUnitsDeprecated; - jvObj[jss::fee_base] = lpAccepted->fees().base.jsonClipped(); - jvObj[jss::reserve_base] = lpAccepted->fees().reserve.jsonClipped(); - jvObj[jss::reserve_inc] = lpAccepted->fees().increment.jsonClipped(); - - jvObj[jss::txn_count] = json::UInt(alpAccepted->size()); - - if (mode_ >= OperatingMode::SYNCING) - { - jvObj[jss::validated_ledgers] = - registry_.get().getLedgerMaster().getCompleteLedgers(); - } - - auto it = streamMaps_[SLedger].begin(); - while (it != streamMaps_[SLedger].end()) - { - InfoSub::pointer const p = it->second.lock(); - if (p) - { - p->send(jvObj, true); - ++it; - } - else - { - it = streamMaps_[SLedger].erase(it); - } - } - } - - if (!streamMaps_[SBookChanges].empty()) - { - json::Value const jvObj = xrpl::RPC::computeBookChanges(lpAccepted); - - auto it = streamMaps_[SBookChanges].begin(); - while (it != streamMaps_[SBookChanges].end()) - { - InfoSub::pointer const p = it->second.lock(); - if (p) - { - p->send(jvObj, true); - ++it; - } - else - { - it = streamMaps_[SBookChanges].erase(it); - } - } - } - - { - static bool kFirstTime = true; - if (kFirstTime) - { - // First validated ledger, start delayed SubAccountHistory - kFirstTime = false; - for (auto& outer : subAccountHistory_) - { - for (auto& inner : outer.second) - { - auto& subInfo = inner.second; - if (subInfo.index->separationLedgerSeq == 0) - { - subAccountHistoryStart(alpAccepted->getLedger(), subInfo); - } - } - } - } - } - } + // Stream updates and the account-history kick-off touch different lock + // domains; each helper takes only its own lock, so the two are never held + // together. + publishLedgerStreams(lpAccepted, alpAccepted); + kickoffAccountHistory(alpAccepted); // Don't lock since pubAcceptedTransaction is locking. for (auto const& accTx : *alpAccepted) { JLOG(journal_.trace()) << "pubAccepted: " << accTx->getJson(); - pubValidatedTransaction(lpAccepted, *accTx, accTx == *(--alpAccepted->end())); + bool const last = &*accTx == &alpAccepted->back(); + pubValidatedTransaction(lpAccepted, *accTx, last); + } +} + +void +NetworkOPsImp::publishLedgerStreams( + std::shared_ptr const& lpAccepted, + std::shared_ptr const& alpAccepted) +{ + // Hold each locked subscriber alive until after streamLock_ is released; a + // last-reference ~InfoSub would otherwise re-acquire this non-recursive + // mutex and self-deadlock. Declared before the lock, destroyed after it; + // covers both the ledger and book-changes loops below. + std::vector toRelease; + + std::scoped_lock const sl(streamLock_); + + if (!streamMaps_[SLedger].empty()) + { + json::Value jvObj(json::ValueType::Object); + + jvObj[jss::type] = "ledgerClosed"; + jvObj[jss::ledger_index] = lpAccepted->header().seq; + jvObj[jss::ledger_hash] = to_string(lpAccepted->header().hash); + jvObj[jss::ledger_time] = + json::Value::UInt(lpAccepted->header().closeTime.time_since_epoch().count()); + + jvObj[jss::network_id] = registry_.get().getNetworkIDService().getNetworkID(); + + if (!lpAccepted->rules().enabled(featureXRPFees)) + jvObj[jss::fee_ref] = kFeeUnitsDeprecated; + jvObj[jss::fee_base] = lpAccepted->fees().base.jsonClipped(); + jvObj[jss::reserve_base] = lpAccepted->fees().reserve.jsonClipped(); + jvObj[jss::reserve_inc] = lpAccepted->fees().increment.jsonClipped(); + + jvObj[jss::txn_count] = json::UInt(alpAccepted->size()); + + if (mode_ >= OperatingMode::SYNCING) + { + jvObj[jss::validated_ledgers] = registry_.get().getLedgerMaster().getCompleteLedgers(); + } + auto it = streamMaps_[SLedger].begin(); + while (it != streamMaps_[SLedger].end()) + { + InfoSub::pointer p = it->second.lock(); + if (p) + { + p->send(jvObj, true); + toRelease.push_back(std::move(p)); + ++it; + } + else + { + it = streamMaps_[SLedger].erase(it); + } + } + } + + if (!streamMaps_[SBookChanges].empty()) + { + json::Value const jvObj = xrpl::RPC::computeBookChanges(lpAccepted); + + auto it = streamMaps_[SBookChanges].begin(); + while (it != streamMaps_[SBookChanges].end()) + { + InfoSub::pointer p = it->second.lock(); + if (p) + { + p->send(jvObj, true); + toRelease.push_back(std::move(p)); + ++it; + } + else + { + it = streamMaps_[SBookChanges].erase(it); + } + } + } +} + +void +NetworkOPsImp::kickoffAccountHistory(std::shared_ptr const& alpAccepted) +{ + // Runs exactly once, the first time a ledger is published. The atomic + // exchange lets the common post-first-ledger path return without taking + // accountLock_, while still admitting exactly one caller even if ledger + // publishing is ever made concurrent. + static std::atomic done{false}; + if (done.exchange(true)) + return; + + // It only reads/writes subAccountHistory_, so it takes accountLock_ alone. + std::scoped_lock const sl(accountLock_); + for (auto& outer : subAccountHistory_) + { + for (auto& inner : outer.second) + { + auto& subInfo = inner.second; + if (subInfo.index->separationLedgerSeq == 0) + subAccountHistoryStart(alpAccepted->getLedger(), subInfo); + } } } @@ -3249,7 +3431,7 @@ NetworkOPsImp::getLocalTxCount() std::size_t NetworkOPsImp::getBookSubscribersCount() { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(bookLock_); std::size_t total = 0; for (auto const& [_, subs] : subBook_) total += subs.size(); @@ -3376,7 +3558,13 @@ NetworkOPsImp::pubValidatedTransaction( MultiApiJson const jvObj = transJson(stTxn, trResult, true, ledger, metaRef); { - std::scoped_lock const sl(subLock_); + // Hold each locked subscriber alive until after streamLock_ is + // released; a last-reference ~InfoSub would otherwise re-acquire this + // non-recursive mutex and self-deadlock. Declared before the lock, + // destroyed after the block ends; covers both loops below. + std::vector toRelease; + + std::scoped_lock const sl(streamLock_); auto it = streamMaps_[STransactions].begin(); while (it != streamMaps_[STransactions].end()) @@ -3388,6 +3576,7 @@ NetworkOPsImp::pubValidatedTransaction( jvObj.visit( p->getApiVersion(), // [&](json::Value const& jv) { p->send(jv, true); }); + toRelease.push_back(std::move(p)); ++it; } else @@ -3407,6 +3596,7 @@ NetworkOPsImp::pubValidatedTransaction( jvObj.visit( p->getApiVersion(), // [&](json::Value const& jv) { p->send(jv, true); }); + toRelease.push_back(std::move(p)); ++it; } else @@ -3431,20 +3621,20 @@ NetworkOPsImp::pubBookTransaction(AcceptedLedgerTx const& alTx, MultiApiJson con // Two-pass design: // - // 1. Under subLock_, walk subBook_, collect a strong pointer for each + // 1. Under bookLock_, walk subBook_, collect a strong pointer for each // unique listener (and prune any expired weak_ptrs we encounter). - // 2. Release subLock_, then send to each collected listener. + // 2. Release bookLock_, then send to each collected listener. // // Reasoning: - // * send() can be slow / blocking, so holding subLock_ across it would - // stall every other sub/unsub/pub path on this server (see the matching - // TODO above pubServer at line ~2275). - // * A strong pointer destructed while subLock_ is held risks running + // * send() can be slow / blocking, so holding bookLock_ across it would + // stall every other book sub/unsub/pub path on this server (see the + // matching TODO above pubServer at line ~2275). + // * A strong pointer destructed while bookLock_ is held risks running // ~InfoSub() in-line, which re-enters unsubBook() and mutates the very // subBook_/SubMapType being iterated -> dangling iterator UB. // - // Releasing subLock_ before any InfoSub::pointer can decay solves both. - // ~InfoSub() reacquires subLock_ via unsubBook() on its own and serializes + // Releasing bookLock_ before any InfoSub::pointer can decay solves both. + // ~InfoSub() reacquires bookLock_ via unsubBook() on its own and serializes // safely with concurrent traffic. std::vector listeners; @@ -3458,7 +3648,7 @@ NetworkOPsImp::pubBookTransaction(AcceptedLedgerTx const& alTx, MultiApiJson con seen.reserve(books.size()); { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(bookLock_); for (auto const& book : books) { @@ -3496,8 +3686,8 @@ NetworkOPsImp::pubBookTransaction(AcceptedLedgerTx const& alTx, MultiApiJson con { jvObj.visit(p->getApiVersion(), [&](json::Value const& jv) { p->send(jv, true); }); } - // listeners destructs here, outside subLock_; ~InfoSub (if any fires) - // will reacquire subLock_ via unsubBook with no iterator hazard. + // listeners destructs here, outside bookLock_; ~InfoSub (if any fires) + // will reacquire bookLock_ via unsubBook with no iterator hazard. } void @@ -3513,7 +3703,7 @@ NetworkOPsImp::pubAccountTransaction( std::vector accountHistoryNotify; auto const currLedgerSeq = ledger->seq(); { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(accountLock_); if (!subAccount_.empty() || !subRTAccount_.empty() || !subAccountHistory_.empty()) { @@ -3646,7 +3836,7 @@ NetworkOPsImp::pubProposedAccountTransaction( std::vector accountHistoryNotify; { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(accountLock_); if (subRTAccount_.empty()) return; @@ -3730,7 +3920,7 @@ NetworkOPsImp::subAccount( isrListener->insertSubAccountInfo(naAccountID, rt); } - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(accountLock_); for (auto const& naAccountID : vnaAccountIDs) { @@ -3773,7 +3963,7 @@ NetworkOPsImp::unsubAccountInternal( hash_set const& vnaAccountIDs, bool rt) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(accountLock_); SubInfoMapType& subMap = rt ? subRTAccount_ : subAccount_; @@ -3795,6 +3985,122 @@ NetworkOPsImp::unsubAccountInternal( } } +template +void +NetworkOPsImp::cleanupSubscriptionMap( + std::uint64_t seq, + hash_set const& accounts, + OuterMap& outerMap, + BeforeErase&& beforeErase) +{ + // Walk the disconnecting connection's accounts in chunks. Each chunk takes + // accountLock_, erases up to kAccountCleanupChunk entries, then releases + // the lock so a competing account-publish can run before the next chunk. + // No iterator into outerMap is held across the unlock: every chunk re-finds + // each account, so a concurrent mutation between chunks cannot dangle. + auto it = accounts.begin(); + auto const end = accounts.end(); + while (it != end) + { + std::scoped_lock const sl(accountLock_); + + for (std::size_t n = 0; n < kAccountCleanupChunk && it != end; ++n, ++it) + { + auto outerIter = outerMap.find(*it); + if (outerIter != outerMap.end()) + { + // Give the caller a chance to tear down this connection's inner + // entry before it is erased (the history map stops its paging + // job here); the plain account maps pass a no-op. + auto innerIter = outerIter->second.find(seq); + if (innerIter != outerIter->second.end()) + beforeErase(innerIter->second); + + // Erase only this connection's seq; other connections sharing + // the account keep their entry, so a reconnect is unaffected. + outerIter->second.erase(seq); + if (outerIter->second.empty()) + outerMap.erase(outerIter); + } + } + } +} + +void +NetworkOPsImp::cleanupAccountSubscriptions( + std::uint64_t seq, + hash_set const& accounts, + SubInfoMapType& subMap) +{ + // Plain account maps need no per-entry teardown before erase. + cleanupSubscriptionMap(seq, accounts, subMap, [](InfoSub::wptr const&) {}); +} + +void +NetworkOPsImp::cleanupAccountHistorySubscriptions( + std::uint64_t seq, + hash_set const& accounts) +{ + // Cancel any in-flight historical paging job for this connection before + // dropping its record. The job holds its own shared_ptr to the index, so + // erasing the map entry alone would not stop it; it reads this atomic + // between pages and exits promptly once set. + cleanupSubscriptionMap( + seq, accounts, subAccountHistory_, [](SubAccountHistoryInfoWeak const& info) { + info.index->stopHistorical = true; + }); +} + +void +NetworkOPsImp::scheduleAccountCleanup( + std::uint64_t seq, + hash_set rtAccounts, + hash_set normalAccounts, + hash_set historyAccounts) +{ + // Nothing to do for a connection that never subscribed to any account. + if (rtAccounts.empty() && normalAccounts.empty() && historyAccounts.empty()) + return; + + // Post the erase work to a low-priority job so the disconnect thread (and + // ~InfoSub) returns immediately. The job captures the sets BY MOVE and + // operates purely on seq + the captured accounts; it never touches the + // destroyed InfoSub. `this` outlives the job per the Source lifetime + // contract. Running on a JobQueue thread, it cannot re-enter accountLock_ + // held by the disconnecting thread, so the plain std::mutex is safe. + // + // The body is exception-guarded: the JobQueue invokes it bare, so an + // escaping exception on the worker thread would terminate the process. + // + // addJob returns false only once the JobQueue has been stopped, i.e. during + // process shutdown. At that point NetworkOPsImp's maps are about to be + // destroyed wholesale and no publish path can run, so dropping the cleanup + // is harmless; no inline fallback is needed. + jobQueue_.addJob( + JtClientAcctHist, + "SubCleanup", + [this, + seq, + rt = std::move(rtAccounts), + normal = std::move(normalAccounts), + history = std::move(historyAccounts)]() noexcept { + try + { + cleanupAccountSubscriptions(seq, rt, subRTAccount_); + cleanupAccountSubscriptions(seq, normal, subAccount_); + cleanupAccountHistorySubscriptions(seq, history); + } + catch (std::exception const& e) + { + JLOG(journal_.error()) << "SubCleanup[seq=" << seq << "]: " << e.what(); + } + catch (...) + { + JLOG(journal_.error()) << "SubCleanup[seq=" << seq << "]: unknown exception"; + } + }); +} + void NetworkOPsImp::addAccountHistoryJob(SubAccountHistoryInfoWeak subInfo) { @@ -4077,7 +4383,7 @@ NetworkOPsImp::subAccountHistory(InfoSub::ref isrListener, AccountID const& acco return RpcInvalidParams; } - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(accountLock_); SubAccountHistoryInfoWeak ahi{ .sinkWptr = isrListener, .index = std::make_shared(accountId)}; auto simIterator = subAccountHistory_.find(accountId); @@ -4125,7 +4431,7 @@ NetworkOPsImp::unsubAccountHistoryInternal( AccountID const& account, bool historyOnly) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(accountLock_); auto simIterator = subAccountHistory_.find(account); if (simIterator != subAccountHistory_.end()) { @@ -4157,7 +4463,7 @@ NetworkOPsImp::subBook(InfoSub::ref isrListener, Book const& book) // prune in pubBookTransaction. With the reverse ordering, ~InfoSub would // call unsubBookInternal for a key that was never inserted server-side. { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(bookLock_); subBook_[book].try_emplace(isrListener->getSeq(), isrListener); } isrListener->insertBookSubscription(book); @@ -4177,7 +4483,7 @@ NetworkOPsImp::unsubBook(InfoSub::ref isrListener, Book const& book) bool NetworkOPsImp::unsubBookInternal(std::uint64_t uSeq, Book const& book) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(bookLock_); auto it = subBook_.find(book); if (it == subBook_.end()) return false; @@ -4227,7 +4533,7 @@ NetworkOPsImp::subLedger(InfoSub::ref isrListener, json::Value& jvResult) jvResult[jss::validated_ledgers] = registry_.get().getLedgerMaster().getCompleteLedgers(); } - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SLedger].emplace(isrListener->getSeq(), isrListener).second; } @@ -4235,7 +4541,7 @@ NetworkOPsImp::subLedger(InfoSub::ref isrListener, json::Value& jvResult) bool NetworkOPsImp::subBookChanges(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SBookChanges].emplace(isrListener->getSeq(), isrListener).second; } @@ -4243,7 +4549,7 @@ NetworkOPsImp::subBookChanges(InfoSub::ref isrListener) bool NetworkOPsImp::unsubLedger(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SLedger].erase(uSeq) != 0u; } @@ -4251,7 +4557,7 @@ NetworkOPsImp::unsubLedger(std::uint64_t uSeq) bool NetworkOPsImp::unsubBookChanges(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SBookChanges].erase(uSeq) != 0u; } @@ -4259,7 +4565,7 @@ NetworkOPsImp::unsubBookChanges(std::uint64_t uSeq) bool NetworkOPsImp::subManifests(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SManifests].emplace(isrListener->getSeq(), isrListener).second; } @@ -4267,7 +4573,7 @@ NetworkOPsImp::subManifests(InfoSub::ref isrListener) bool NetworkOPsImp::unsubManifests(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SManifests].erase(uSeq) != 0u; } @@ -4292,7 +4598,7 @@ NetworkOPsImp::subServer(InfoSub::ref isrListener, json::Value& jvResult, bool a jvResult[jss::pubkey_node] = toBase58(TokenType::NodePublic, registry_.get().getApp().nodeIdentity().first); - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SServer].emplace(isrListener->getSeq(), isrListener).second; } @@ -4300,7 +4606,7 @@ NetworkOPsImp::subServer(InfoSub::ref isrListener, json::Value& jvResult, bool a bool NetworkOPsImp::unsubServer(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SServer].erase(uSeq) != 0u; } @@ -4308,7 +4614,7 @@ NetworkOPsImp::unsubServer(std::uint64_t uSeq) bool NetworkOPsImp::subTransactions(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[STransactions].emplace(isrListener->getSeq(), isrListener).second; } @@ -4316,7 +4622,7 @@ NetworkOPsImp::subTransactions(InfoSub::ref isrListener) bool NetworkOPsImp::unsubTransactions(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[STransactions].erase(uSeq) != 0u; } @@ -4324,7 +4630,7 @@ NetworkOPsImp::unsubTransactions(std::uint64_t uSeq) bool NetworkOPsImp::subRTTransactions(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SRtTransactions].emplace(isrListener->getSeq(), isrListener).second; } @@ -4332,7 +4638,7 @@ NetworkOPsImp::subRTTransactions(InfoSub::ref isrListener) bool NetworkOPsImp::unsubRTTransactions(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SRtTransactions].erase(uSeq) != 0u; } @@ -4340,7 +4646,7 @@ NetworkOPsImp::unsubRTTransactions(std::uint64_t uSeq) bool NetworkOPsImp::subValidations(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SValidations].emplace(isrListener->getSeq(), isrListener).second; } @@ -4354,7 +4660,7 @@ NetworkOPsImp::stateAccounting(json::Value& obj) bool NetworkOPsImp::unsubValidations(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SValidations].erase(uSeq) != 0u; } @@ -4362,7 +4668,7 @@ NetworkOPsImp::unsubValidations(std::uint64_t uSeq) bool NetworkOPsImp::subPeerStatus(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SPeerStatus].emplace(isrListener->getSeq(), isrListener).second; } @@ -4370,7 +4676,7 @@ NetworkOPsImp::subPeerStatus(InfoSub::ref isrListener) bool NetworkOPsImp::unsubPeerStatus(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SPeerStatus].erase(uSeq) != 0u; } @@ -4378,7 +4684,7 @@ NetworkOPsImp::unsubPeerStatus(std::uint64_t uSeq) bool NetworkOPsImp::subConsensus(InfoSub::ref isrListener) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SConsensusPhase].emplace(isrListener->getSeq(), isrListener).second; } @@ -4386,15 +4692,14 @@ NetworkOPsImp::subConsensus(InfoSub::ref isrListener) bool NetworkOPsImp::unsubConsensus(std::uint64_t uSeq) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); return streamMaps_[SConsensusPhase].erase(uSeq) != 0u; } InfoSub::pointer -NetworkOPsImp::findRpcSub(std::string const& strUrl) +NetworkOPsImp::findRpcSubLocked(std::string const& strUrl) { - std::scoped_lock const sl(subLock_); - + // Caller already holds streamLock_; this performs the lookup only. auto const it = rpcSubMap_.find(strUrl); if (it != rpcSubMap_.end()) @@ -4403,10 +4708,17 @@ NetworkOPsImp::findRpcSub(std::string const& strUrl) return InfoSub::pointer(); } +InfoSub::pointer +NetworkOPsImp::findRpcSub(std::string const& strUrl) +{ + std::scoped_lock const sl(streamLock_); + return findRpcSubLocked(strUrl); +} + InfoSub::pointer NetworkOPsImp::addRpcSub(std::string const& strUrl, InfoSub::ref rspEntry) { - std::scoped_lock const sl(subLock_); + std::scoped_lock const sl(streamLock_); rpcSubMap_.emplace(strUrl, rspEntry); @@ -4416,20 +4728,31 @@ NetworkOPsImp::addRpcSub(std::string const& strUrl, InfoSub::ref rspEntry) bool NetworkOPsImp::tryRemoveRpcSub(std::string const& strUrl) { - std::scoped_lock const sl(subLock_); - auto pInfo = findRpcSub(strUrl); - - if (!pInfo) - return false; - - // check to see if any of the stream maps still hold a weak reference to - // this entry before removing - for (SubMapType const& map : streamMaps_) + // Declared before the lock so it outlives the scoped_lock and is destroyed + // only after streamLock_ is released. The erase below may drop the last + // strong reference; if so, ~InfoSub runs and its unsub* calls re-acquire + // the non-recursive streamLock_. Destroying pInfo inside the lock would + // self-deadlock. + InfoSub::pointer pInfo; { - if (map.contains(pInfo->getSeq())) + std::scoped_lock const sl(streamLock_); + // Use the no-lock helper: we already hold streamLock_ and the mutex is + // not recursive, so calling the public findRpcSub here would deadlock. + pInfo = findRpcSubLocked(strUrl); + + if (!pInfo) return false; + + // check to see if any of the stream maps still hold a weak reference to + // this entry before removing + for (SubMapType const& map : streamMaps_) + { + if (map.contains(pInfo->getSeq())) + return false; + } + rpcSubMap_.erase(strUrl); } - rpcSubMap_.erase(strUrl); + // pInfo destroyed here, after streamLock_ is released. return true; } diff --git a/src/xrpld/core/Config.h b/src/xrpld/core/Config.h index 852e46218a..a7cb5d053a 100644 --- a/src/xrpld/core/Config.h +++ b/src/xrpld/core/Config.h @@ -229,6 +229,12 @@ public: static constexpr int kMaxJobQueueTx = 1000; static constexpr int kMinJobQueueTx = 100; + // Optional override for the per-connection subscription cap. Unset means + // use the built-in default (kMaxSubscriptionsPerConnection in InfoSub.h). + // Kept as an override here, rather than the default itself, so the core + // module need not depend on the server module that owns the constant. + std::optional maxSubscriptionsPerConnection; + // Amendment majority time std::chrono::seconds amendmentMajorityTime = kDefaultAmendmentMajorityTime; diff --git a/src/xrpld/core/detail/Config.cpp b/src/xrpld/core/detail/Config.cpp index 3b7b57328b..616717c5fd 100644 --- a/src/xrpld/core/detail/Config.cpp +++ b/src/xrpld/core/detail/Config.cpp @@ -677,6 +677,9 @@ Config::loadFromString(std::string const& fileContents) if (getSingleSection(secConfig, Sections::kNetworkQuorum, strTemp, j_)) networkQuorum = beast::lexicalCastThrow(strTemp); + if (getSingleSection(secConfig, Sections::kMaxSubscriptionsPerConnection, strTemp, j_)) + maxSubscriptionsPerConnection = beast::lexicalCastThrow(strTemp); + fees = setupFeeVote(section(Sections::kVoting)); /* [fee_default] is documented in the example config files as useful for * things like offline transaction signing. Until that's completely diff --git a/src/xrpld/rpc/handlers/subscribe/Subscribe.cpp b/src/xrpld/rpc/handlers/subscribe/Subscribe.cpp index 93840bb6d6..cf43501c52 100644 --- a/src/xrpld/rpc/handlers/subscribe/Subscribe.cpp +++ b/src/xrpld/rpc/handlers/subscribe/Subscribe.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include #include @@ -19,6 +20,7 @@ #include #include +#include #include #include #include @@ -26,6 +28,24 @@ namespace xrpl { +namespace { + +/** + * Test whether admitting `additional` subscriptions would exceed the cap. + * + * @param ispSub The connection's InfoSub, queried for its current count. + * @param additional Number of new items this branch would add. + * @param cap The effective per-connection cap for this request. + * @return true if the request must be rejected to stay within the cap. + */ +[[nodiscard]] bool +wouldExceedSubscriptionCap(InfoSub::ref ispSub, std::size_t additional, std::size_t cap) +{ + return exceedsSubscriptionCap(ispSub->totalSubscriptionCount(), additional, cap); +} + +} // namespace + json::Value doSubscribe(RPC::JsonContext& context) { @@ -105,6 +125,11 @@ doSubscribe(RPC::JsonContext& context) } ispSub->setApiVersion(context.apiVersion); + // Effective per-connection subscription cap: a configured override if set, + // otherwise the built-in default. Resolved once and reused by every branch. + std::size_t const subscriptionCap = + context.app.config().maxSubscriptionsPerConnection.value_or(kMaxSubscriptionsPerConnection); + if (context.params.isMember(jss::streams)) { if (!context.params[jss::streams].isArray()) @@ -166,30 +191,59 @@ doSubscribe(RPC::JsonContext& context) } } + // Parse the proposed (real-time) and normal account sets first, then check + // the cap against their COMBINED net-new total before subscribing either. + // This keeps the account pair all-or-nothing: it never subscribes one set + // and then rejects on the other. Other fields (streams and account_history) + // are still checked and subscribed independently, as they always have been, + // so a later field can be rejected after an earlier one subscribed. The cap + // counts only NET-NEW accounts (those not already tracked on this + // connection), so re-subscribing accounts already held is never wrongly + // rejected. auto accountsProposed = context.params.isMember(jss::accounts_proposed) ? jss::accounts_proposed : jss::rt_accounts; // DEPRECATED - if (context.params.isMember(accountsProposed)) + bool const hasProposed = context.params.isMember(accountsProposed); + bool const hasAccounts = context.params.isMember(jss::accounts); + + hash_set proposedIds; + hash_set accountIds; + + if (hasProposed) { if (!context.params[accountsProposed].isArray()) return rpcError(RpcInvalidParams); - auto ids = RPC::parseAccountIds(context.params[accountsProposed]); - if (ids.empty()) + proposedIds = RPC::parseAccountIds(context.params[accountsProposed]); + if (proposedIds.empty()) return rpcError(RpcActMalformed); - context.netOps.subAccount(ispSub, ids, true); } - if (context.params.isMember(jss::accounts)) + if (hasAccounts) { if (!context.params[jss::accounts].isArray()) return rpcError(RpcInvalidParams); - auto ids = RPC::parseAccountIds(context.params[jss::accounts]); - if (ids.empty()) + accountIds = RPC::parseAccountIds(context.params[jss::accounts]); + if (accountIds.empty()) return rpcError(RpcActMalformed); - context.netOps.subAccount(ispSub, ids, false); - JLOG(context.j.debug()) << "doSubscribe: accounts: " << ids.size(); + } + + if (hasProposed || hasAccounts) + { + // Atomic check-and-reserve, so two concurrent requests sharing this + // InfoSub (admin subscribe-by-url) cannot both pass the cap check. + if (!ispSub->tryReserveAccountSubscriptions(proposedIds, accountIds, subscriptionCap)) + return RPC::makeParamError("Too many subscriptions for this connection."); + } + + if (hasProposed) + context.netOps.subAccount(ispSub, proposedIds, true); + + if (hasAccounts) + { + context.netOps.subAccount(ispSub, accountIds, false); + JLOG(context.j.debug()) << "doSubscribe: accounts: " << accountIds.size(); } if (context.params.isMember(jss::account_history_tx_stream)) @@ -206,6 +260,13 @@ doSubscribe(RPC::JsonContext& context) if (!id) return rpcError(RpcInvalidParams); + // Charge the cap only when net-new, like the account branches. Not + // atomic here (subAccountHistory does its own dup-detecting insert), but + // a concurrent race adds at most one entry, so the overshoot is trivial. + std::size_t const historyCharge = ispSub->hasAccountHistorySubscription(*id) ? 0 : 1; + if (wouldExceedSubscriptionCap(ispSub, historyCharge, subscriptionCap)) + return RPC::makeParamError("Too many subscriptions for this connection."); + if (auto result = context.netOps.subAccountHistory(ispSub, *id); result != RpcSuccess) { return rpcError(result); @@ -222,6 +283,10 @@ doSubscribe(RPC::JsonContext& context) if (!context.params[jss::books].isArray()) return rpcError(RpcInvalidParams); + // Book subscriptions are tracked separately (OrderBookDB) and are not + // part of totalSubscriptionCount(), so they are not gated by the + // per-connection account cap. Each book entry is validated and + // subscribed below. for (auto& j : context.params[jss::books]) { if (!j.isObject() || !j.isMember(jss::taker_pays) || !j.isMember(jss::taker_gets) ||