Fix hangs in RPC clients against a dying/terminated rpc-server

[hpk42]

Startup hang seen on flaky job run for PR #8601 CI (unrelated change),
https://github.com/chatmail/core/actions/runs/31965959724/job/95211728194

Excerpt

tests/test_something.py:747:
.tox/py/lib/pypy3.10/site-packages/deltachat_rpc_client/rpc.py:121: in start
    system_info = self.get_system_info()
.tox/py/lib/pypy3.10/site-packages/deltachat_rpc_client/rpc.py:46: in rpc_future
    response = queue.get()
E   Failed: Timeout (>300.0s) from pytest-timeout.

FAILED tests/test_something.py::test_early_failure - Failed: Timeout (>300.0s...
============= 1 failed, 114 passed, 1 skipped in 408.11s (0:06:48) =============

This Python RPC client hang illustrates a race, introduced with the startup health-check (cb783ff), by yours truly. Rpc.start() spawns the reader/writer/events threads and then issues get_system_info(), which registers its result queue in request_results. When the server exits right away, the reader loop hits EOF and its cleanup unblocks only already-registered requests. A request registered after that is never answered, so its queue.get() blocks forever.

The race window is the server's startup-to-exit time, on my machine 5-10 ms. Delaying registration by 10 ms reproduces the hang 10/10. The test change added in this PR showcases the problem deterministically. The same race affects any RPC call issued while the server dies, not just the startup health-check from (cb783ff). It also turned out that a failed start() never winds down its threads. The non-daemon writer thread stays parked on the request queue, so a plain non-xdist pytest tests/test_something.py::test_early_failure passes and then can hangs at interpreter exit.

The PR makes request registration synchronize with reader-loop shutdown: once the reader loop has finished, new requests immediately raise "RPC server closed" instead of waiting for a response that can no longer arrive. A failed start() additionally closes stdin and joins the loop threads, so they cannot block interpreter exit.

Closes #8511

[link2xt]

Possibly closes #8511

[hpk42]

@link2xt addressed all your comments and force-pushed, am resolving your comments now to make fresh room for more :)

The lock is gone (although i think it read more easily, to be honest), and there is a shared _shutdown_loops helper now. Turned out that stdin.close() is not safe on the failed start path: if the writer's flush already hit EPIPE then stdin.close() retries the flush and raises, so we need to suppress BrokenPipeError.

All of this btw reminds me of the unnerving struggles around execnet-teardown, see e.g. https://github.com/pytest-dev/execnet/blob/607ba0b29857705eb3f7f47f3953d40d72b96a96/src/execnet/gateway_base.py#L1055-L1061
(which does use a lock FWIW)

[link2xt]

The lock is gone (although i think it read more easily, to be honest)

Conceptually (code-wise you still need to figure out what the bool does etc. maybe) reasoning about channels/pipes/queues is easier than locks, because i can think about the writer and reader part separately. On the writer side, try to write the request into the queue, and if it fails (because the queue is closed) you need to serve the request yourself by returning an error. On the reader side, read the requests and answer them, and then when you want to shutdown, close the writer side of the pipe ("shutdown"), process the rest of the requests. It is easy to check that no matter how many writers you have, each request is processed by some process, either the reader, or the writer failing to put the request into a pipe.

With the lock code to figure out if it does the right thing, i need to find all the places where the lock is used, then think about three threads (at least, reader and two writers) starting right before the lock and think through all the orders in which they may acquire the locks. Or i can reduce it to queues and think about readers/writers separately.

[link2xt]

There is another events_thread.join() inside of _shutdown_loops(), do we still need this join here? Also self.closing = True can be moved inside of _shutdown_loops, it is a signal for events_thread to stop.

[hpk42]

the join in close() is about ordering: it lets the events_loop stop on the closing flag before the pipe is torn down. Dropping it (with moving closing into _shutdown_loops) would work and never hang, but the events loop might call get_next_event_batch() and exit via RPC server closed, which then sprinkles ERROR ... Exception in the event loop. The double events_thread.join is a no-op for close so i think it's fine, and i added a comment to explain the ordering constraint.