proxy: add local connection limit to ListenConnections

[enirox001]

This adds an optional local connection limit toListenConnections().

Previously, ListenConnections() would accept incoming connections indefinitely. This branch adds an optional max_connections parameter so a listener can stop accepting new connections once a per-listener cap is reached, and resume accepting when an existing connection disconnects.

The limit is local to the listener instead of global to the EventLoop. This keeps the state and behavior scoped to the listening socket, and is closer to the direction discussed downstream for per--ipcbind limits.

This also adds a test covering the behavior with max_connections=1, verifying that:

• the first client is accepted normally
• a second client is not accepted while the first remains connected
• the second client is accepted after the first disconnects

Note This PR includes a major version bump to v12 due to the API addition. If #274 lands earlier and bumps the version to v12 first, we will need to bump the version here again.

[DrahtBot]

The following sections might be updated with supplementary metadata relevant to reviewers and maintainers.

Reviews

See the guideline and AI policy for information on the review process.

Type	Reviewers
ACK	ryanofsky
Stale ACK	xyzconstant, Eunovo

If your review is incorrectly listed, please copy-paste <!--meta-tag:bot-skip--> into the comment that the bot should ignore.

Conflicts

No conflicts as of last run.

[ryanofsky]

Approach ACK 84ed607. Implementation of local connection limit here looks almost exactly like I would have expected.

I do think it would be helpful to see a draft PR in the bitcoin repo using this API (it should be fine to make libmultiprocess changes there and let the lint CI job fail so you don't need to mess with subtrees) because the approach in bitcoin/bitcoin#34978 of adding a global connection limit option isn't exactly compatible with the implementation here of implementing a per-address connection limit.

I'd personally prefer using per-address limits over introducing a global limit but both approaches seem reasonable

[ryanofsky]

In commit "proxy: add local connection limit to ListenConnections" (84ed607)

Curious if this accept_pending variable is actually necessary or if code could just compare active_connections and max_connections when deciding whether to listen. Would prefer to avoid redundancy in the state representation if possible even if makes individual checks more a little more verbose.

If accept_pending really is necessary would be a good to have a short comment about why.

[enirox001]

I kept accept_pending, but added a short comment explaining why it is needed here. active_connections only counts accepted connections, so without a separate flag nested _Listen() calls could post multiple pending accept() calls before active_connections is incremented.

[xyzconstant]

FWIW I tested the code without accept_pending field and hence its check at _Listen removed, the tests passed with no issues. Then, asked Claude to generate a test that exercises it and produced this:

diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp
index b367938..78298c7 100644
--- a/test/mp/test/listen_tests.cpp
+++ b/test/mp/test/listen_tests.cpp
@@ -24,8 +24,10 @@
 #include <string>
 #include <sys/socket.h>
 #include <sys/un.h>
+#include <kj/exception.h>
 #include <thread>
 #include <unistd.h>
+#include <vector>
 
 namespace mp {
 namespace test {
@@ -112,11 +114,39 @@ public:
     std::thread thread;
 };
 
+//! kj::ExceptionCallback that captures KJ_LOG output into an external sink.
+//! Must be instantiated on the thread whose KJ logs you want to capture; it
+//! installs itself onto that thread's ExceptionCallback stack via its base
+//! constructor and removes itself in the destructor.
+class CaptureLogCallback : public kj::ExceptionCallback
+{
+public:
+    CaptureLogCallback(std::mutex& mu, std::string& sink) : m_mu(mu), m_sink(sink) {}
+
+    void logMessage(kj::LogSeverity severity, const char* file, int line, int contextDepth,
+                    kj::String&& text) override
+    {
+        {
+            std::lock_guard<std::mutex> lock(m_mu);
+            m_sink.append(text.cStr(), text.size());
+            m_sink.push_back('\n');
+        }
+        // Still let the default callback emit to stderr so test debug output
+        // isn't silenced for other observers.
+        kj::ExceptionCallback::logMessage(severity, file, line, contextDepth, kj::mv(text));
+    }
+
+private:
+    std::mutex& m_mu;
+    std::string& m_sink;
+};
+
 class ListenSetup
 {
 public:
     explicit ListenSetup(std::optional<size_t> max_connections = std::nullopt)
         : capped_listener(max_connections.has_value()), thread([this, max_connections] {
+              CaptureLogCallback log_capture(captured_log_mutex, captured_log);
               EventLoop loop("mptest-server", [this](mp::LogMessage log) {
                   if (log.level == mp::Log::Raise) throw std::runtime_error(log.message);
                   if (log.message.find("IPC server: socket connected.") != std::string::npos) {
@@ -144,6 +174,18 @@ public:
 
     ~ListenSetup()
     {
+        forceShutdown();
+        thread.join();
+    }
+
+    //! Synchronously tear down the event loop's task set so any pending accept
+    //! promises are destroyed now (rather than when the destructor runs later).
+    //! This makes it possible to assert on captured KJ log output before the
+    //! ListenSetup goes out of scope. Idempotent.
+    void forceShutdown()
+    {
+        if (shutdown_done) return;
+        shutdown_done = true;
         if (capped_listener) {
             EventLoop* loop;
             {
@@ -152,7 +194,6 @@ public:
             }
             if (loop) loop->sync([&] { loop->m_task_set.reset(); });
         }
-        thread.join();
     }
 
     size_t ConnectedCount()
@@ -184,11 +225,15 @@ public:
     UnixListener listener;
     std::promise<void> ready_promise;
     bool capped_listener{false};
+    bool shutdown_done{false};
     std::mutex counter_mutex;
     std::condition_variable counter_cv;
     EventLoop* event_loop{nullptr};
     size_t connected_count{0};
     size_t disconnected_count{0};
+    //! KJ log output captured from the server thread via CaptureLogCallback.
+    std::mutex captured_log_mutex;
+    std::string captured_log;
     std::thread thread;
 };
 
@@ -245,6 +290,34 @@ KJ_TEST("ListenConnections keeps capped listeners alive before reaching the limi
     KJ_EXPECT(client2->client->add(2, 3) == 5);
 }
 
+// Without `accept_pending`, cascaded close handlers each post a duplicate
+// accept(). KJ silently serializes them so the cap isn't exceeded, but the
+// extra pending promises are destroyed at cleanup and logged as
+// "PromiseFulfiller was destroyed without fulfilling the promise."
+// This test fails when accept_pending is removed and passes when it's intact.
+KJ_TEST("ListenConnections does not leak accept promises during disconnect burst")
+{
+    constexpr size_t kCap = 2;
+    ListenSetup setup(/*max_connections=*/kCap);
+
+    std::vector<std::unique_ptr<ClientSetup>> filling;
+    filling.reserve(kCap);
+    for (size_t i = 0; i < kCap; ++i) {
+        filling.push_back(std::make_unique<ClientSetup>(setup.listener.Connect()));
+    }
+    setup.WaitForConnectedCount(kCap);
+
+    filling.clear();
+    setup.WaitForDisconnectedCount(kCap);
+
+    // Trigger m_task_set.reset() now so any leaked accept promises get destroyed
+    // before we read captured_log.
+    setup.forceShutdown();
+
+    std::lock_guard<std::mutex> lock(setup.captured_log_mutex);
+    KJ_EXPECT(setup.captured_log.find("PromiseFulfiller was destroyed") == std::string::npos);
+}
+
 } // namespace
 } // namespace test
 } // namespace mp

Basically what this does is install a kj::ExceptionCallback in the server thread to capture the log "PromiseFulfiller was destroyed" generated by KJ runtime if cascaded disconnects each call _Listen and post a fresh accept() promise. The test fails with accept_pending removed and pass with it back.

[xyzconstant]

IMO it's not so obvious why this field is needed here, this patch basically guarantee the same outcome and also pass the test generated by Claude above:

diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h
index 78924c6..c002811 100644
--- a/include/mp/proxy-io.h
+++ b/include/mp/proxy-io.h
@@ -851,11 +851,6 @@ struct ListenState
     kj::Own<kj::ConnectionReceiver> listener;
     std::optional<size_t> max_connections;
     size_t active_connections{0};
-    //! Tracks whether accept() has already been posted. This is needed because
-    //! active_connections only counts accepted connections, so without a
-    //! separate flag, nested _Listen() calls could queue multiple pending
-    //! accepts before active_connections increases.
-    bool accept_pending{false};
 };
 
 template <typename InitInterface, typename InitImpl>
@@ -866,9 +861,12 @@ void _ServeAccepted(EventLoop& loop, InitImpl& init, const std::shared_ptr<Liste
 {
     ++state->active_connections;
     _Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, state] {
+        const bool was_at_cap = state->max_connections && state->active_connections == *state->max_connections;
         assert(state->active_connections > 0);
         --state->active_connections;
-        _Listen<InitInterface>(loop, init, state);
+        if (was_at_cap) {
+            _Listen<InitInterface>(loop, init, state);
+        }
     });
 }
 
@@ -885,15 +883,12 @@ inline std::unique_ptr<EventLoopRef> _MakeCappedListenerRef(EventLoop& loop, con
 template <typename InitInterface, typename InitImpl>
 void _Listen(EventLoop& loop, InitImpl& init, const std::shared_ptr<ListenState>& state)
 {
-    if (state->accept_pending) return;
     if (_ListenAtCapacity(*state)) return;
 
-    state->accept_pending = true;
     auto* ptr = state->listener.get();
     auto accept_ref{_MakeCappedListenerRef(loop, *state)};
     loop.m_task_set->add(ptr->accept().then(
         [&loop, &init, state, accept_ref = std::move(accept_ref)](kj::Own<kj::AsyncIoStream>&& stream) mutable {
-            state->accept_pending = false;
             _ServeAccepted<InitInterface>(loop, init, state, kj::mv(stream));
             _Listen<InitInterface>(loop, init, state);
         }));

[enirox001]

I think this is a better invariant than tracking accept_pending.

There should already be one pending accept whenever the capped listener is below capacity, and disconnects only need to post a new accept when they transition the listener from full to below full, because that is the only state where nocaccept was pending.

So checking this before decrementing active_connections avoids the duplicate-accept case without adding another state variable.

Taken and decided to use resume_accept for the local boolean instead

[enirox001]

Thanks for the review @ryanofsky .

I addressed the cleanup points in the latest push:

• _Serve is now explicit at all call sites
• accept_pending is documented,
• listener coverage now lives in a dedicated listen_tests.cpp file introduced in a separate commit before the local-limit extension.

I’m also planning to put together a draft Bitcoin Core PR using this API so the per-address approach can be evaluated downstream against the current global-limit direction.

[enirox001]

I put together the downstream draft using this API here bitcoin/bitcoin#35037

It uses per--ipcbind max-connections=<n> options, threads the parsed per-address limit into ListenConnections(), and adds downstream coverage for the local-limit behavior.

[enirox001]

This PR is now ready for review

[xyzconstant]

Code review ACK 8511c68

Reviewed each commit separately, compiled and ran tests. The changes look good to me, only left a couple of inline nits noting a compilation error + failing tests in the first commit.

[xyzconstant]

nit: This will fail to compile mptest at commit 8c47a3a since the 4th parameter max_connections is not introduced in ListenConnections's signature until next commit. I'd fix the call for this commit and then adjust the setup accordingly in 8511c68.

[enirox001]

Fixed, thanks

[xyzconstant]

nit: Unlike in ListenConnections's call issue (compilation error), this will still fail for this commit (8c47a3a). Following ryanosfky's reasoning (#269 (comment)) I believe this suite would be better introduced in 8511c68 so the test lands with the feature it exercises.

[enirox001]

Done

[enirox001]

Thanks for the review @xyzconstant . Fixed both commit-structure issues: the first test commit now only adds baseline ListenConnections() coverage using the existing 3-argument API, and the max_connections setup/test now lands with the feature commit that introduces the new parameter

[enirox001]

Also tightened the capped listener behavior. It now stops posting accepts once the limit is reached, and keeps capped pending accepts alive so later clients can connect after an idle gap, including before the cap has been reached.

Added coverage for the reconnect cases as well

[xyzconstant]

In commit "proxy: add local connection limit to ListenConnections" (19e1386)

Not sure if I'm missing something but this check here seems redundant with the same _ListenAtCapacity check at the start of _Listen (line 889).

I tested commenting this line out and left the upper-level check (and vice-versa) and the tests passed with no issues. I'd suggest dropping any of these duplicates.

[enirox001]

Good catch, thanks. Dropped the lower _ListenAtCapacity() check since _Listen() already handles the capacity check before posting another accept.

[xyzconstant]

With the accept_pending check in place, this _Listen call will never be reached.

NOTE: if you apply this patch here (comment), then it will be needed here because we can't rely on the second _Listen call in _ServeAccepted which is executed conditionally after active_connections reached the cap.

[xyzconstant]

Thanks for the update @enirox001!

I've been playing around with the PR code more throughly this time and have a different take on accept_pending now, so left a few comments.

Overall the code is factually correct and works as expected. And I think it could be merged (despite my latest thoughts on accept_pending). So I'll just re-ACK b0207dd

[enirox001]

This looks good in it's current form and could be merged. Let me know if you'd prefer that or want to make another round of updates

Thanks for giving this a look, i intend to make another round of updates. Should be able to have that done soon

[enirox001]

Thanks for the reviews @ryanofsky @Eunovo. Addressed them in the latest commits

I'm also planning to open a new bitcoin core PR bumping the subtree and incorporating this change after this PR is merged

This would be helpful to bitcoin/bitcoin#35037 as well as greatly simplify the work needed for the PR. Looking forward to reviewing this

[ryanofsky]

Code review ACK f9ef92e. Thanks for the updates!Feel free to ignore my comment about whitespace. I only posted it since I noticed CI was failing and thought there would have to be another push

[enirox001]

Thanks for the updates!Feel free to ignore my comment about whitespace. I only posted it since I noticed CI was failing and thought there would have to be another push

Addressed these in the recent change. Faced some IWYU errors after the comment above. So had to resolve them anyway. Thanks

[ryanofsky]

Code review ACK 39a10ce

Just include and whitespace changes since last review

[xyzconstant]

Post-merge re-ACK 39a10ce