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 for information on the review process.

Type	Reviewers
ACK	Eunovo, ryanofsky
Stale ACK	xyzconstant

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

Conflicts

Reviewers, this pull request conflicts with the following ones:

• #298 (WIP: Add test coverage for ConnectStream by xyzconstant)

If you consider this pull request important, please also help to review the conflicting pull requests. Ideally, start with the one that should be merged first.

[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

[ryanofsky]

Code review e86184b. Sorry again for the long delay reviewing this and I will try to be more responsive. This PR seems easier to understand with the new refactoring, so thanks for making that change. I suggested another simplification below which I think would be important to make, but overall the change looks very good.

[ryanofsky]

In commit "proxy: add local connection limit to ListenConnections" (8d09ad3)

I understand point of sleep above is to make sure KJ_EXPECT(setup.ConnectedCount() == 1); does not pass accidentally due to timing even if max_connections implementation is broken.

But I don't understand reason this sleep is useful. Would be good to add a comment if it is helpful, or drop otherwise

[enirox001]

This indeed has no use here, dropped this sleep since the following WaitForConnectedCount(3) already verifies that accepting resumed.

[ryanofsky]

re: #269 (comment)

This indeed has no use here, dropped this sleep since the following WaitForConnectedCount(3) already verifies that accepting resumed.

I seem to still see the sleep in db81e9c (line 223)

[enirox001]

Replaced this sleep with (**setup.m_loop_ref).sync([] {}) with the comment above it explaining why is is used as well

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (3547a26)

I think you could replace this sleep with a call to (**m_loop_ref).sync([] {}); or similar. Test seems to fail reliably with sync instead of sleep when max_connections is increased, so I think the test is still ensures the limit is enforced. Would also suggest a comment for the sleep/sync like "Without this delay, ConnectedCount() == 1 might pass even max_connections was not enforced"

[enirox001]

Done. replaced the sleep with (**setup.m_loop_ref).sync([] {}) and added a comment explaining that the sync gives the event loop a chance to accept the second client if max_connections is not being enforced.

[ryanofsky]

re: #269 (comment)

Done. replaced the sleep with (**setup.m_loop_ref).sync([] {}) and added a comment explaining that the sync gives the event loop a chance to accept the second client if max_connections is not being enforced.

I see this is added in a new test but existing sleep seems to remain in db81e9c line 212

[enirox001]

Removed this sleep entirely.

The following WaitForConnectedCount(3) already verifies that accepting resumed after the previous client disconnected, so no extra delay is needed here.

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (3547a26)

This test just seems to be testing what happens when there is a connection, a disconnect, and then another connect serially so do does not seem to be cover anything the previous test doesn't cover. It might make more sense for this test to create two connections at the same time and make sure both work, but that a third connection doesn't work. This way there is coverage of connections working in parallel with the limit.

[enirox001]

Added this test in db81e9c

It verifies that when two clients are connected and served concurrently with max_connections=2, a third client is not accepted while the listener is at capacity, and that the third client is accepted only after one active client disconnects.

[ryanofsky]

re: #269 (comment)

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

Added this test in db81e9c

Thanks for adding the new test, but suggestion was really to expand the "ListenConnections keeps capped listeners alive before reaching the limit" test to do this, because this test doesn't seem to test anything new that the previous "enforces a local connection limit" test doesn't cover.

So now would suggest keeping the "enforces" test and "parallel connections" tests but dropping the "capped listeners" test.

[enirox001]

but suggestion was really to expand the "ListenConnections keeps capped listeners alive before reaching the limit" test to do this, because this test doesn't seem to test anything new that the previous "enforces a local connection limit" test doesn't cover.

After adding the "multiple connections" test, kept this test as well because i thought it would be nice to have the added coverage.

But like you mentioned. This is a redundant test that doesn't cover any distinct functionality. I have removed it.

[enirox001]

Thanks for the detailed review @ryanofsky

But I'd tweak the commit message to drop "before bumping the version to v12" because when v11 and v11.0 tags are created they will need to point to last merge commit before this PR. So they won't actually include these release notes.

Reworded the commit message to avoid implying the v11 notes are included in this PR’s release. It now says the v11 notes describe changes that will be tagged before this PR.

Also made the changes requested in the latest push.

I dropped the mpgen IWYU commit because it was adding includes to generated files that IWYU reported as unused. The actual IWYU issue was that listen_tests.cpp uses EventLoopRef directly, so I fixed that by including mp/proxy.h in listen_tests.cpp

[Eunovo]

Concept ACK db81e9c:

Left some comments

[Eunovo]

db81e9c

Looks like this same pattern works for connected_count, and we don't need the testing_hook_connected:

if (log.message.find("IPC server: socket connected.") != std::string::npos) {
    std::lock_guard<std::mutex> lock(counter_mutex);
    ++connected_count;
    counter_cv.notify_all();
}

[Sjors]

I think a testing hook is cleaner than parsing log strings.

[Eunovo]

I think a testing hook is cleaner than parsing log strings.

Indeed, but I prefer that we don't add testing hooks unnecessarily. Besides, he is already parsing the disconnected log string; might as well parse the connected log string.

[ryanofsky]

re: #269 (comment)

Indeed, but I prefer that we don't add testing hooks unnecessarily.

Would be curious to if there's any particular reason why. I feel like they add very little overhead and have similar side-benefits as log messages in making code more readable and grep-able. I would definitely not shy away from them in cases where they can add better test coverage and more compile time safety than log string checks.

[enirox001]

I understand why we would want to parse the log string instead of the testing hook, but i do not think there is much downside to adding a testing hook.

I think it makes the code much cleaner, and is lean enough that it wouldn't cause much problems by using it

I will keep the connected hook because it avoids making the test depend on log message text to make the test consistent and avoid string parsing entirely,

Also added a disconnected_hook and switched disconnected_count to use that instead.

[Eunovo]

SGTM

[Eunovo]

922bfba:
I think we can merge ClientSetup and ListenSetup into one TestSetup class. The clients and listener can use the same EventLoop instance.

[Sjors]

I'm not sure if that makes things easier to follow? I would expect the server to have its own EventLoop.

[Eunovo]

I'm not sure if that makes things easier to follow?

It's not necessary, but it's less code.

I would expect the server to have its own EventLoop.

Indeed, but it does not matter in this test. We can have less code if they shared the same EventLoop. The same pattern is used in existing test setup https://github.com/bitcoin-core/libmultiprocess/blob/master/test/mp/test/test.cpp#L64

I'm angling for less code overall, but this works too.

[ryanofsky]

re: #269 (comment)

I'd agree with with suggestion to reduce code duplication and share one event loop. I think having one eventloop in tests is generally better because it adds creates more tasks and dependencies for the async code to keep track of and is more likely to break if things don't happen in the right order.

But I also think it's good to have some variety in tests and not always use one event loop, so either way seem ok.

[enirox001]

We can have less code if they shared the same EventLoop. The same pattern is used in existing test setup https://github.com/bitcoin-core/libmultiprocess/blob/master/test/mp/test/test.cpp#L64

I think this is a good idea, and I spent some time thinking about it and playing around with an implementation. But after trying it, I think using the same pattern as TestSetup makes this fixture harder to reason about.

The existing TestSetup pattern works well for in-memory client to server tests, where the test is mostly checking serialization and application behavior like “if I call add(1, 2), does it reach the server and return 3?”

These ListenConnections tests are different because they are exercising a real listening socket and multiple independent clients connecting to it. For that, I think keeping ClientSetup and ListenSetup separate makes the client and listener roles explicit and keeps the test closer to the behavior being tested which is a listener running on one side, and clients connecting from outside.

I agree this is a little more code, but I think the separation buys clarity here. So I’d prefer to keep this structure unless there’s a concrete simplification that preserves that separation.

[Eunovo]

db81e9c:

The connections in this test look completely synchronous to me. A better simulation of simultaneous connection requests can be created with std::thread and std::barrier. I also think it's better to make >2 simultaneous requests and test that only 2 were accepted.

You can also test simultaneous disconnection. Although it should be noted that these tests will pass because the kernel will queue the connection and disconnection events, and even if it didn't, Listener::listen processes the connection requests one at a time, and the active connections count is decremented on the event loop, which should make it "thread-safe".

[ryanofsky]

re: #269 (comment)

db81e9c:

A better simulation of simultaneous connection requests can be created with [...]

These are good ideas. Might be better to change the current test name to refer "multiple" connections instead of "parallel", since the connections are established and calls are made serially. I suggested this test in #269 (comment) and I think having it is better than not having it, but it could be extended to do more.

[enirox001]

Might be better to change the current test name to refer "multiple" connections instead of "parallel", since the connections are established and calls are made serially

Renamed the test to refer to multiple active connections instead of parallel connections. agree “parallel” overstated what it was doing because the connection setup and calls are serial.

A better simulation of simultaneous connection requests can be created with std::thread and std::barrier. I also think it's better to make >2 simultaneous requests and test that only 2 were accepted.

I’m a bit averse to adding a barrier-based simultaneous-connect test right now because I think it would mostly exercise kernel socket backlog behavior and event loop scheduling.

Listener::listen still processes accepted streams on the event loop one at a time, and m_active_connections is updated on that same even -loop thread.

So the existing test covers the important listener behavior which is multiple active connections are allowed up to the configured limit, a further connection is not accepted while at capacity, and accepting resumes after a disconnect.

[Eunovo]

db81e9c:

I'm not sure what this test is trying to achieve; can you explain?

[ryanofsky]

re: #269 (comment)

I'm not sure what this test is trying to achieve; can you explain?

I think this test is a holdover from before the suggested lifetime simplification #269 (comment) and I suggested deleting in #269 (comment)

[enirox001]

I'm not sure what this test is trying to achieve; can you explain?

As mentioned above by @ryanofsky this test was added when the initial listeners functionality were added. But this has been simplified, but it seems i had not removed it.

This has been removed in dcf9c38

[Sjors]

Feedback on the first commit.

[Sjors]

In 922bfba test: add dedicated ListenConnections coverage: it would be good to briefly document the ListenSetup, ClientSetup and UnixListener classes - and their relationship.

[enirox001]

Done. Added short comments in ef35679 explaining the functionality as well as the relationships of these classes.

In the same vein, i have added some comments to some of the non trivial tests in listen_tests.cpp file. This should make it easier to understand and follow

[Sjors]

In 922bfba test: add dedicated ListenConnections coverage: I think it's better to introduce m_loop_ref here rather than in the next commit. I found myself wondering why the loop doesn't immediately self-destruct before the first client connects. It doesn't, for subtle reasons. Explicitly taking a reference makes it more clear, especially combined with #302.

diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp
index 86e6421777..9629ce69d0 100644
--- a/test/mp/test/listen_tests.cpp
+++ b/test/mp/test/listen_tests.cpp
@@ -18,6 +18,8 @@
 #include <kj/test.h>
 #include <memory>
+#include <mp/proxy.h>
 #include <mp/proxy-io.h>
 #include <mutex>
+#include <optional>
 #include <ratio> // IWYU pragma: keep
 #include <stdexcept>
@@ -131,4 +133,5 @@ public:
                   counter_cv.notify_all();
               };
+              m_loop_ref.emplace(loop);
               FooImplementation foo;
               ListenConnections<messages::FooInterface>(loop, listener.release(), foo);
@@ -142,4 +145,5 @@ public:
     ~ListenSetup()
     {
+        m_loop_ref.reset();
         thread.join();
     }
@@ -157,4 +161,5 @@ public:
     UnixListener listener;
     std::promise<void> ready_promise;
+    std::optional<EventLoopRef> m_loop_ref;
     std::mutex counter_mutex;
     std::condition_variable counter_cv;

[enirox001]

Good point. I have moved the m_loop_ref setup into the initial listen_tests.cpp commit in ef35679 so the listener event loop lifetime is explicit from the start.

The subsequent max_connections tests still use the same m_loop_ref for event-loop synchronization, but the commit now owns the reference before those tests are introduced

[Sjors]

In 922bfba test: add dedicated ListenConnections coverage: maybe call it server, so the next line is easier to read.

[enirox001]

Done, renames this to server.

This pattern is repeated in the subsequent commit, so i also renamed them to server as well. This makes the tests easier to read and understand

[Sjors]

In 922bfba test: add dedicated ListenConnections coverage: maybe rename to MakeConnectedSocket()

[enirox001]

Done. Renamed UnixListener::Connect() to MakeConnectedSocket() to make it clearer that the helper returns a connected socket file descriptor.

[ryanofsky]

Code review ACK db81e9c. Looks good! Main change is simplifying lifetime code and dropping m_listeners list since last review. Would be good to respond to remaining comments though.

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (922bfba)

This is fine for this PR, but would be interesting for a followup to see if this could be replaced using libkj IO functions to listen on a random TCP port or unix socket path in a more portable way that could work on windows. Might allow code to be simplified too

[enirox001]

Noted. Seeing that this is open #274 to add non unix support. This is something i intend to look into

Using libkj IO helpers for a more portable temporary listener setup seems a good starting point, especially if it can also reduce the socket setup code here.

[ryanofsky]

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

Just want to note for followup that it will probably make sense to return this listener shared_ptr to the caller to allow it to stop listening. Right now there isn't (and has never been) a way to stop listening without shutting down the entire event loop, but this could now be supported with the listener object. (see also #269 (comment))

[enirox001]

Having a way to make a stop a listener from listening is something that should definitiely be supported. And returning the std::shared_ptr<Listener> later as a way to potentially do that seems like a good starting point.

[ryanofsky]

re: #269 (comment)

This indeed has no use here, dropped this sleep since the following WaitForConnectedCount(3) already verifies that accepting resumed.

I seem to still see the sleep in db81e9c (line 223)

[ryanofsky]

re: #269 (comment)

Done. replaced the sleep with (**setup.m_loop_ref).sync([] {}) and added a comment explaining that the sync gives the event loop a chance to accept the second client if max_connections is not being enforced.

I see this is added in a new test but existing sleep seems to remain in db81e9c line 212

[ryanofsky]

re: #269 (comment)

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

Added this test in db81e9c

Thanks for adding the new test, but suggestion was really to expand the "ListenConnections keeps capped listeners alive before reaching the limit" test to do this, because this test doesn't seem to test anything new that the previous "enforces a local connection limit" test doesn't cover.

So now would suggest keeping the "enforces" test and "parallel connections" tests but dropping the "capped listeners" test.

[ryanofsky]

re: #269 (comment)

Indeed, but I prefer that we don't add testing hooks unnecessarily.

Would be curious to if there's any particular reason why. I feel like they add very little overhead and have similar side-benefits as log messages in making code more readable and grep-able. I would definitely not shy away from them in cases where they can add better test coverage and more compile time safety than log string checks.

[ryanofsky]

re: #269 (comment)

I'd agree with with suggestion to reduce code duplication and share one event loop. I think having one eventloop in tests is generally better because it adds creates more tasks and dependencies for the async code to keep track of and is more likely to break if things don't happen in the right order.

But I also think it's good to have some variety in tests and not always use one event loop, so either way seem ok.

[ryanofsky]

re: #269 (comment)

I'm not sure what this test is trying to achieve; can you explain?

I think this test is a holdover from before the suggested lifetime simplification #269 (comment) and I suggested deleting in #269 (comment)

[ryanofsky]

re: #269 (comment)

db81e9c:

A better simulation of simultaneous connection requests can be created with [...]

These are good ideas. Might be better to change the current test name to refer "multiple" connections instead of "parallel", since the connections are established and calls are made serially. I suggested this test in #269 (comment) and I think having it is better than not having it, but it could be extended to do more.

[enirox001]

Thanks for all the reviews so far @Eunovo @Sjors @ryanofsky . Addressed all the comments in the latest commits.

[Eunovo]

ACK dcf9c38.

[Eunovo]

ef35679:

Not necessarily important because this is test code, but you can assert m_fd >= 0 before releasing here.

[Eunovo]

dcf9c38:

I think you meant // Without this sync, ConnectedCount() == 1 might pass even if

[Eunovo]

dcf9c38:

// client2 yet

[ryanofsky]

Code review ACK dcf9c38.

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. I'm also planning to open a new bitcoin core PR bumping the subtree and incorporating this change after this is merged.

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (ef35679)

Would change "server thread" to "event loop thread". Saying server thread is a little confusing because typically there's only event loop and it processes all events on a single thread, not distinguishing between client and server events.

Would also change "the client" to "a client" since there may be more than one.

Same suggestion also applies to next commit adding disconnect hook.

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (ef35679)

Hardcoding /tmp is probably fine because this is just creating a socket, but it would be a little better to respect TMPDIR variable, maybe using std::filesystem::temp_directory_path() iike bitcoin tests

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (ef35679)

Would be a little better to use Mutex and Lock classes from util.h since they have thread safety annotations.

Same comment applies to new locks and mutexes added in the next commit.

(As possible followup it might be nice to have a linter that disallows non-annotated classes by default)

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (ef35679)

Unexplained hardcoded timeouts in tests like this make code more difficult to understand and maintain. Would suggest defining constexpr auto FAILURE_TIMEOUT{30s}; similar to spawn_tests. The constant can then be used here and elsewhere for similar operations that are expected to complete quickly but will cause the test to fail if they do not.

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (ef35679)

I think it's a little confusing and creates an opportunity for bugs that a non-static method is taking a self parameter, and could potentially be called with two different Listener instances.

Also, if we want to start returning Listener objects from ListenConnections to applications to stop listening, it would be confusing for this method to be part of the public interface, since applications should never call it.

For both of these reasons would suggest dropping the Listen::listener method, and just keeping the previous _Listen function which is only meant to be used internally and not called by applications. This would also be a code simplification

Suggested change:

diff

--- a/include/mp/proxy-io.h
+++ b/include/mp/proxy-io.h
@@ -870,32 +870,27 @@ struct Listener
         return m_max_connections && m_active_connections >= *m_max_connections;
     }
 
-    //! Handle incoming connections by calling _Serve, to create ProxyServer
-    //! objects and forward requests to the init object.
-    template <typename InitInterface, typename InitImpl>
-    void listen(EventLoop& loop, InitImpl& init, const std::shared_ptr<Listener>& self);
-
     kj::Own<kj::ConnectionReceiver> m_receiver;
     std::optional<size_t> m_max_connections;
     size_t m_active_connections{0};
 };
 
 template <typename InitInterface, typename InitImpl>
-void Listener::listen(EventLoop& loop, InitImpl& init, const std::shared_ptr<Listener>& self)
+void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImpl& init)
 {
-    if (atCapacity()) return;
+    if (listener->atCapacity()) return;
 
-    auto* receiver = m_receiver.get();
+    auto* receiver = listener->m_receiver.get();
     loop.m_task_set->add(receiver->accept().then(
-        [&loop, &init, self](kj::Own<kj::AsyncIoStream>&& stream) {
-            ++self->m_active_connections;
-            _Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, self] {
-                const bool resume_accept{self->atCapacity()};
-                assert(self->m_active_connections > 0);
-                --self->m_active_connections;
-                if (resume_accept) self->listen<InitInterface>(loop, init, self);
+        [&loop, &init, listener](kj::Own<kj::AsyncIoStream>&& stream) {
+            ++listener->m_active_connections;
+            _Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, listener] {
+                const bool resume_accept{listener->atCapacity()};
+                assert(listener->m_active_connections > 0);
+                --listener->m_active_connections;
+                if (resume_accept) _Listen<InitInterface>(listener, loop, init);
             });
-            self->listen<InitInterface>(loop, init, self);
+            _Listen<InitInterface>(listener, loop, init);
         }));
 }
 
@@ -920,7 +915,7 @@ void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<si
         auto listener{std::make_shared<Listener>(
             loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
             max_connections)};
-        listener->listen<InitInterface>(loop, init, listener);
+        _Listen<InitInterface>(listener, loop, init);
     });
 }
 

[ryanofsky]

In commit "test: add dedicated ListenConnections coverage" (ef35679)

May want to declare this explicit now since it becomes explicit anyway in the next commit

[ryanofsky]

In commit "doc/version: Bump version 11 > 12" (2f7f1bf)

These updates look right but might better to use the more complete list of changes from bitcoin/bitcoin#35661. Suggested diff is below, but also feel free to keep current test or just replace the list with a TBD comment.

diff

--- a/doc/versions.md
+++ b/doc/versions.md
@@ -14,11 +14,17 @@ include.
   and resume accepting after existing connections disconnect.
 
 ## [v11.0](https://github.com/bitcoin-core/libmultiprocess/commits/v11.0)
-- Tolerates unexpected exceptions in event loop `post()` callbacks.
-- Tolerates exceptions from remote destroy during cleanup in `ProxyClient`.
-- Supports primitive `std::optional` struct fields in the code generator (`mpgen`).
-- Adds `TypeName()` and improves debug log coverage for Proxy object lifecycle.
-- Updates build compatibility with recent Nix and CMake versions.
+- Adds `makePool` method on `ThreadMap` to support thread pool routing, allowing requests without a specific client thread to be dispatched to a pool using a shortest-queue strategy ([#283](https://github.com/bitcoin-core/libmultiprocess/pull/283)).
+- Adds `std::unordered_set` support, a `BuildList` helper, and a `ReadList` helper to reduce duplication in list build and read handlers ([#277](https://github.com/bitcoin-core/libmultiprocess/pull/277), [#285](https://github.com/bitcoin-core/libmultiprocess/pull/285)).
+- Adds support for translating C++ `std::optional<T>` struct fields to pairs of `T` + `hasT :Bool` Cap'n Proto struct fields, allowing unset optional primitive fields to be represented ([#243](https://github.com/bitcoin-core/libmultiprocess/pull/243)).
+- Produces more readable log output for Proxy object lifecycle events and IPC server-side failures ([#218](https://github.com/bitcoin-core/libmultiprocess/pull/218)).
+- Handles exceptions thrown by `destroy` methods by logging instead of aborting ([#273](https://github.com/bitcoin-core/libmultiprocess/pull/273)). This can prevent server crashes when non-libmultiprocess clients disconnect without destroying objects, in the case where a server object owns client objects and the server destructor tries to call the disconnected client to free them ([#219](https://github.com/bitcoin-core/libmultiprocess/issues/219)).
+- Handles unexpected exceptions thrown by callbacks (that should never happen) by logging errors instead of deadlocking ([#260](https://github.com/bitcoin-core/libmultiprocess/pull/260)).
+- Fixes a rare mptest hang on musl builds caused by a lost wakeup bug in `Waiter` ([#295](https://github.com/bitcoin-core/libmultiprocess/pull/295)).
+- Fixes a race condition in a log print detected by TSan ([#286](https://github.com/bitcoin-core/libmultiprocess/pull/286)).
+- Build improvements: makes `target_capnp_sources` work correctly when libmultiprocess is used as a CMake subproject ([#289](https://github.com/bitcoin-core/libmultiprocess/pull/289)), adds `mp_headers` target for better lint tool support ([#291](https://github.com/bitcoin-core/libmultiprocess/pull/291)), and fixes compatibility with recent Nix and CMake 4.0 ([#238](https://github.com/bitcoin-core/libmultiprocess/pull/238)).
+- Test, CI, documentation, and minor code improvements: design document corrections ([#278](https://github.com/bitcoin-core/libmultiprocess/pull/278)), field constant comments ([#279](https://github.com/bitcoin-core/libmultiprocess/pull/279)), clang-tidy fix ([#292](https://github.com/bitcoin-core/libmultiprocess/pull/292)), new smoke test for double-precision float values ([#294](https://github.com/bitcoin-core/libmultiprocess/pull/294)), new test for recursive async IPC calls ([#301](https://github.com/bitcoin-core/libmultiprocess/pull/301)), removal of libevent from Core CI builds ([#299](https://github.com/bitcoin-core/libmultiprocess/pull/299)), and rename of `EventLoop::m_num_clients` to `m_num_refs` ([#302](https://github.com/bitcoin-core/libmultiprocess/pull/302)).
+- Used in Bitcoin Core master branch, pulled in by [#35661](https://github.com/bitcoin/bitcoin/pull/35661).
 
 ## [v10.0](https://github.com/bitcoin-core/libmultiprocess/commits/v10.0)
 - Increases spawn test timeout to avoid spurious failures.

[ryanofsky]

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

I think it would make sense before each WaitFor{Connected,Disconnected}Count call to have a KJ_EXPECT call checking the previous value. Should make the test stronger and also clearer and easier to debug if something is wrong

diff

--- a/test/mp/test/listen_tests.cpp
+++ b/test/mp/test/listen_tests.cpp
@@ -168,6 +168,12 @@ public:
         return connected_count;
     }
 
+    size_t DisconnectedCount()
+    {
+        std::lock_guard<std::mutex> lock(counter_mutex);
+        return disconnected_count;
+    }
+
     void WaitForConnectedCount(size_t expected_count)
     {
         std::unique_lock<std::mutex> lock(counter_mutex);
@@ -203,8 +209,8 @@ public:
 KJ_TEST("ListenConnections accepts incoming connections")
 {
     ListenSetup server;
+    KJ_EXPECT(server.ConnectedCount() == 0);
     auto client = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
-
     server.WaitForConnectedCount(1);
     KJ_EXPECT(client->client->add(1, 2) == 3);
 }
@@ -217,29 +223,34 @@ KJ_TEST("ListenConnections enforces a local connection limit")
 
     ListenSetup server(/*max_connections=*/1);
 
+    KJ_EXPECT(server.ConnectedCount() == 0);
     auto client1 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
     server.WaitForConnectedCount(1);
+
     KJ_EXPECT(client1->client->add(1, 2) == 3);
 
     auto client2 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
-
     // Without this sync, ConnectedCount() == 2 might pass even if
     // max_connections was not enforced because the event loop has not accepted
     // client3 yet.
     (**server.m_loop_ref).sync([] {});
-    KJ_EXPECT(server.ConnectedCount() == 1);
 
+    KJ_EXPECT(server.ConnectedCount() == 1);
+    KJ_EXPECT(server.DisconnectedCount() == 0);
     client1.reset();
     server.WaitForDisconnectedCount(1);
     server.WaitForConnectedCount(2);
 
     KJ_EXPECT(client2->client->add(2, 3) == 5);
 
+    KJ_EXPECT(server.DisconnectedCount() == 1);
     client2.reset();
     server.WaitForDisconnectedCount(2);
 
+    KJ_EXPECT(server.ConnectedCount() == 2);
     auto client3 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
     server.WaitForConnectedCount(3);
+
     KJ_EXPECT(client3->client->add(3, 4) == 7);
 }
 
@@ -250,22 +261,22 @@ KJ_TEST("ListenConnections accepts multiple connections")
     
     ListenSetup server(/*max_connections=*/2);
 
+    KJ_EXPECT(server.ConnectedCount() == 0);
     auto client1 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
     auto client2 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
-
     server.WaitForConnectedCount(2);
 
     KJ_EXPECT(client1->client->add(1, 2) == 3);
     KJ_EXPECT(client2->client->add(2, 3) == 5);
 
     auto client3 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
-
     // Without this sync, ConnectedCount() == 2 might pass even if
     // max_connections was not enforced because the event loop has not accepted
     // client3 yet.
     (**server.m_loop_ref).sync([] {});
-    KJ_EXPECT(server.ConnectedCount() == 2);
 
+    KJ_EXPECT(server.ConnectedCount() == 2);
+    KJ_EXPECT(server.DisconnectedCount() == 0);
     client1.reset();
     server.WaitForDisconnectedCount(1);
     server.WaitForConnectedCount(3);