Avoid errors from asynchronous (non-c++) clients

[ryanofsky]

The PR avoids errors from non-C++ rust & python clients when they make asynchronous requests (bitcoin/bitcoin#33923) and unclean disconnects (bitcoin/bitcoin#34250). Neither of these errors are easily possible to trigger from libmultiprocess clients because of its blocking interface and RAII semantics, but they are fairly easy to trigger from rust and python and there is a test triggering both of them in bitcoin/bitcoin#34284

[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
Concept ACK	ismaelsadeeq

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:

• #218 (Better error and log messages by ryanofsky)

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.

LLM Linter (✨ experimental)

Possible typos and grammar issues:

• is is -> is [duplication: "is" is repeated ("m_thread_ready is is signaled")]
• letting parameters to be deleted -> letting parameters be deleted [incorrect infinitive usage; use "be deleted" instead of "to be deleted"]

^{2026-02-11 20:07:08}

[ryanofsky]

Updated ba865d9 -> 92a2c5f (pr/promise.1 -> pr/promise.2, compare) fixing various ci errors (iwyu, compiler errors) https://github.com/bitcoin-core/libmultiprocess/actions/runs/21196838571

[ismaelsadeeq]

Concept ACK

[ryanofsky]

Updated 92a2c5f -> 1f307fb (pr/promise.2 -> pr/promise.3, compare) fixing tsan error caused by deleting ready_fulfiller promise fulfiller outside the event loop thread https://github.com/bitcoin-core/libmultiprocess/actions/runs/21233968144/job/61097851891?pr=240

[ryanofsky]

Have been debugging a TSAN race condition (https://github.com/bitcoin/bitcoin/actions/runs/21421194207/job/61680781122?pr=34422#step:11:4303) the functional test in bitcoin/bitcoin#34422 detects in the new code, that is not triggered here because this PR does not currently include a test that triggers IPC request cancellation.

The race seems real and fixable, also probably unlikely to cause problems in practice. In the control flow when an IPC gets cancelled, TSAN sees the worker thread doing a CallContext::getParams call before the cancellation, triggering a read of the RpcCallContext::request pointer. Then after the cancellation occurs, the I/O thread wakes up and destroys the CallContext object, triggering a write of RpcCallContext::request pointer. This is not actually dangerous because the write always happens before the read, but TSAN is unable to detect any "happens-before" relationship between the read and write events, so considers this to be a race.

I think this should be fixable by having the worker thread acquire a lock in the brief period when it reads request parameters, before executing the IPC call, and also acquire the lock after executing the call when it is writing request results. Then the cancellation callback, if triggered, can acquire the same lock and TSAN can be aware of the ordering between the two threads

[Sjors]

should be fixable by having the worker thread acquire a lock in the brief period ...

@ryanofsky is that something you want to add here?

[ryanofsky]

re: #240 (comment)

should be fixable by having the worker thread acquire a lock in the brief period ...

@ryanofsky is that something you want to add here?

Yes thanks, added a new commit which does this and fixes the TSAN error locally for me. Will update bitcoin/bitcoin#34422 and confirm it fixes the failure there too.

---

Rebased 1f307fb -> 1d7debf (pr/promise.3 -> pr/promise.4, compare) adding result_value.reset() call to try fix CI error https://github.com/bitcoin/bitcoin/actions/runs/21412348994/job/61652239914?pr=34422

Updated 1d7debf -> 4f42fc4 (pr/promise.4 -> pr/promise.5, compare) to fix CI error https://github.com/bitcoin/bitcoin/actions/runs/21421194207/job/61680781122?pr=34422

[theuni]

@ryanofsky I'm having a bit of trouble understanding what's happening on which thread, but it seems like a mutex is not the most straightforward solution here? It's not immediately obvious to me what it's guarding, anyway. Perhaps a simple std::atomic_flag would be better to communicate "I've reached this point"? You could set/notify when the params are safe to delete, then in m_on_cancel you could just wait for it?

[ryanofsky]

@ryanofsky I'm having a bit of trouble understanding what's happening on which thread, but it seems like a mutex is not the most straightforward solution here? It's not immediately obvious to me what it's guarding, anyway. Perhaps a simple std::atomic_flag would be better to communicate "I've reached this point"? You could set/notify when the params are safe to delete, then in m_on_cancel you could just wait for it?

Thanks for looking at this! I assume you are asking about the last commit 4f42fc4 which is fixing the race condition detected by TSAN.

The issue being fixed in that commit is a race between two threads both potentially accessing capnproto request parameters in an incoming request.

• One thread is a worker thread running the big invoke lambda which is responsible for converting the capnproto request parameters into c++ arguments, and then calling a c++ method with those arguments, and then converting the c++ method's return value & output arguments or exception into a capnproto response.
• The other thread is the capnproto event loop thread running the on_cancel lambda when there is a disconnect or cancellation. After the on_cancel lambda returns, capnproto deletes the request parameters.

The point of the mutex is to prevent the event loop thread from deleting the request parameters while the worker thread is still reading them.

The control flow is a little complicated because of the way parameters are unpacked recursively, but the idea of guarding access to the parameters with a mutex is pretty simple. The params_mutex is locked by the worker thread before it accesses the capnproto parameters, and unlocked by that thread after all the parameters have been unpacked and it is ready to call the c++ method that is supposed to be executed. Then on the event loop thread inside the on_cancel lambda, the same mutex is locked before it returns so if there is a cancellation while the worker thread is unpacking parameters, the event loop thread will be blocked, and not delete the parameters until after the worker thread has finished reading them.

What is odd about this fix is that the event loop thread is releasing the mutex before it actually deletes the parameters. This is suspicious, because if the mutex is not locked while they are deleted it does nothing to solve the reverse race condition in the case where the cancellation happens before the parameters start being read. But this case is handled by a separate cancel_monitor.m_cancelled check which raises an exception to prevent the worker thread from reading the parameters.

So the mutex should solve delete-while-reading race condition in a fairly straightforward way even if the control flow is complicated by a lot of recursion and callbacks, and even if it doesn't solve the related delete-before-reading race conditions.

Tangent: One observation looking at the code is that while the delete-before-reading reverse case is being handled, it isn't being handled very well because the exception raised is thrown inside the sync block instead of outside of it, and it should probably be an InterruptException instead of an MP_LOG call so it can be caught higher in the stack and the thread can be cleaned up. I'm guessing right now that code will cause the worker thread to hang be leaked, never getting cleaned up. Ideally I' could write a unit test trying to trigger this, but so far I've been struggling to figure out how to test different possible races and make sure they are handled well. It's not obvious how to arrange the the request being cancelled before parameters are read. Anyway, this is only indirectly related to the TSAN race addressed in the commit, which is also unlikely to happen in practice and probably hasn't happened, even though TSAN was right to warn the request parameters were previously being accessed from different threads without synchronization between them.

[theuni]

@ryanofsky Thanks for the explanation, that was very helpful.

I'm still struggling to understand how this mutex is working though. It seems that since it's already been unlocked in invoke(), the mutex will already be unlocked on the worker thread when params_lock goes out of scope, which would cause a double unlock (undefined), no?

[Sjors]

I created a branch and had an agent splinter 558ae11 refactor: Add ProxyServer::post() method into multiple tiny commits: sjors/2026/02/558ae11-split

No need to use it here, but it might aid others in reviewing (with dimmed-zebra). Noticed a small behavior change in that it removes MP_LOG for thread busy. It's harmless since 92fed35 drops this exception anyway.

Also drew a little diagram to orient myself about where this function lives (IIUC):

Review placeholder: the first two commits, up to 92fed35, mostly make sense to me now.

[theuni]

I'm still struggling to understand how this mutex is working though. It seems that since it's already been unlocked in invoke(), the mutex will already be unlocked on the worker thread when params_lock goes out of scope, which would cause a double unlock (undefined), no?

Nevermind, I see my mistake now. I was thinking it was the bare mutex that was being unlocked in invoke(), but it's the lock itself. Seems sane I suppose, just not exactly straightforward.

[Sjors]

In 92fed35 Allow simultaneous calls on same Context.thread: maybe add a comment that the goal is to add requested fn() calls to the promise chain as quickly as possible, so we call ready_fulfiller->fulfill(); before executing it.

[Sjors]

The third commit looks good to me as well.

[Sjors]

In 0a22425 Prevent crash on unclean disconnect if abandoned IPC call returns interface pointer: for waitNext() the time spent in this race-vulnerable state is minuscule compared to the multi second wait?

In that case, let's not deal with it in this PR.

[Sjors]

In 0a22425 Prevent crash on unclean disconnect if abandoned IPC call returns interface pointer: can you add a comment here to explain why it's important to call sync() without doing anything in particular (where !erase_thread)?

IIUC it's to ensure InterruptException gets thrown as needed, so we can catch it below.

No test breaks (for this commit) if I do:

diff --git a/include/mp/type-context.h b/include/mp/type-context.h
index 134542e..440a3ef 100644
--- a/include/mp/type-context.h
+++ b/include/mp/type-context.h
@@ -113,5 +113,5 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     // makes another IPC call), so avoid modifying the map.
                     const bool erase_thread{inserted};
-                    KJ_DEFER(
+                    KJ_DEFER(if (erase_thread) {
                         // Erase the request_threads entry on the event loop
                         // thread with loop->sync(), so if the connection is
@@ -121,5 +121,4 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                         server.m_context.loop->sync([&] {
                             auto self_dispose{kj::mv(self)};
-                            if (erase_thread) {
                             // Look up the thread again without using existing
                             // iterator since entry may no longer be there after
@@ -133,7 +132,6 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                                 removed = request_threads.extract(server.m_context.connection);
                             }
-                            }
                         });
-                    );
+                    });
                     KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]{
                         try {

(but I guess that makes sense because the commit doesn't touch tests in general)

[ismaelsadeeq]

reviewed the first two commits to 92fed35

Thanks @Sjors for your split of the first commit, and the diagram, it was sweet to use that in the review.

[ismaelsadeeq]

In "Allow simultaneous calls on same Context.thread" 92fed35

Before this commit, ProxyServer<Thread>::post() threw immediately when the thread was busy:

After 92fed35, post() queues the call by chaining onto m_thread_ready instead of throwing, which enables pipelining but removes the only backpressure mechanism. There is now no bound on chain depth. Every call from a fast producer appends a new node to the chain, each holding a captured lambda with serialized arguments plus two kj::Promise objects; the chain grows without limit.

I'm curious whether it's something we should worry about in practice as a potential for OOM of the server, either by bug or deliberate?

I modified the existing test to simulate a slow server (10s sleep per request) with a large number of queued calls:

 setup.server->m_impl->m_int_fn = [&](int n) {
+    std::this_thread::sleep_for(std::chrono::seconds{10});
     assert(n == expected);
     expected += 100;
     return n;
 };

-std::atomic<size_t> running{3};
+std::atomic<size_t> running{1000001};

The mptest process consumed 17 GB RSS in under 30 seconds.

With while(true), the mptest was killed by OOM:

Out of memory: Killed process 2461509 (mptest)
total-vm:53,439,684kB anon-rss:53,617,500kB

Worth noting that in the mptest process the client and server share the same address space, so the 17 GB figure includes client-side memory (unsent/pending capnp requests) as well as server-side promise chain growth?

A potential naive solution, if this is deemed worth fixing, is a depth counter with a limit. increment before queuing, throw if the limit is exceeded, and decrement after the result is dispatched

+    uint64_t m_maximum_promise_depth{1000};
+    uint64_t m_pending_tasks{0};

 template<typename T, typename Fn>
 kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
 {
+    if (m_pending_tasks >= m_maximum_promise_depth) {
+        throw std::runtime_error("maximum promise depth reached");
+    }
+    m_pending_tasks += 1;
+
     auto ready = kj::newPromiseAndFulfiller<void>();
     auto ret = m_thread_ready.then([this, fn = std::forward<Fn>(fn),
                                     ready_fulfiller = kj::mv(ready.fulfiller)]() mutable {
         ...
-        m_loop->sync([&result_value, &exception, result_fulfiller = kj::mv(result_fulfiller)]() mutable {
+        m_loop->sync([this, &result_value, &exception, result_fulfiller = kj::mv(result_fulfiller)]() mutable {
             KJ_IF_MAYBE(e, exception) {
                 result_fulfiller->reject(kj::mv(*e));
             } else {
                 result_fulfiller->fulfill(kj::mv(*result_value));
                 result_value.reset();
             }
+            m_pending_tasks--;
             result_fulfiller = nullptr;
         });
     });

With the fix applied, running mptest with a modified running=1001, a 10s server sleep holds RSS at a stable 17 MB and throws cleanly at depth 1000:

remote exception: std::exception: maximum promise depth reached
VmRSS: 17,484 kB

At running=1000001 with the fix, RSS reached ~9.5 GB from client memory alone.

This implies that in a real deployment where client and server are separate processes, the server would be protected by this fix, but the client could still exhaust its own memory independently. This fix addresses the server-side only.

Maybe there is even a more interesting approach with kj?

[ismaelsadeeq]

In "refactor: Add ProxyServer::post() method" 558ae11

nit: clarify that we return a T promise?

[ismaelsadeeq]

In "refactor: Add ProxyServer::post() method" 558ae11

Are we passing the whole connection here for future profing, i.e we need more things apart from m_loop later?