feat: allow to connect a python rpc-client to an rpc-server via a unix FIFO file instead of subprocess

Splits the RPC client into a transport-agnostic BaseRpc plus RpcSubprocess
(aliased as Rpc for backwards compatibility) and the new RpcFIFO.
Also strikes unneccessary `**kwargs`.

deltachat-rpc-client/src/deltachat_rpc_client/__init__.py

@@ -8,7 +8,7 @@ from .const import EventType, SpecialContactId
 from .contact import Contact
 from .deltachat import DeltaChat
 from .message import Message
-from .rpc import JsonRpcError, Rpc
+from .rpc import JsonRpcError, Rpc, RpcFIFO
 
 __all__ = [
     "Account",
@@ -23,6 +23,7 @@ __all__ = [
     "Message",
     "SpecialContactId",
     "Rpc",
+    "RpcFIFO",
     "run_bot_cli",
     "run_client_cli",
 ]

deltachat-rpc-client/src/deltachat_rpc_client/_utils.py

@@ -53,7 +53,6 @@ def run_client_cli(
     hooks: Optional[Iterable[Tuple[Callable, Union[type, "EventFilter"]]]] = None,
     until: Callable[[AttrDict], bool] = _forever,
     argv: Optional[list] = None,
-    **kwargs,
 ) -> None:
     """Run a simple command line app, using the given hooks.
 
@@ -61,14 +60,13 @@ def run_client_cli(
     """
     from .client import Client
 
-    _run_cli(Client, until, hooks, argv, **kwargs)
+    _run_cli(Client, until, hooks, argv)
 
 
 def run_bot_cli(
     until: Callable[[AttrDict], bool] = _forever,
     hooks: Optional[Iterable[Tuple[Callable, Union[type, "EventFilter"]]]] = None,
     argv: Optional[list] = None,
-    **kwargs,
 ) -> None:
     """Run a simple bot command line using the given hooks.
 
@@ -76,7 +74,7 @@ def run_bot_cli(
     """
     from .client import Bot
 
-    _run_cli(Bot, until, hooks, argv, **kwargs)
+    _run_cli(Bot, until, hooks, argv)
 
 
 def _run_cli(
@@ -84,7 +82,6 @@ def _run_cli(
     until: Callable[[AttrDict], bool] = _forever,
     hooks: Optional[Iterable[Tuple[Callable, Union[type, "EventFilter"]]]] = None,
     argv: Optional[list] = None,
-    **kwargs,
 ) -> None:
     from .deltachat import DeltaChat
     from .rpc import Rpc
@@ -102,7 +99,7 @@ def _run_cli(
     parser.add_argument("--password", action="store", help="password", default=os.getenv("DELTACHAT_PASSWORD"))
     args = parser.parse_args(argv[1:])
 
-    with Rpc(accounts_dir=args.accounts_dir, **kwargs) as rpc:
+    with Rpc(accounts_dir=args.accounts_dir) as rpc:
         deltachat = DeltaChat(rpc)
         core_version = (deltachat.get_system_info()).deltachat_core_version
         accounts = deltachat.get_all_accounts()

deltachat-rpc-client/src/deltachat_rpc_client/rpc.py

@@ -10,17 +10,21 @@ import subprocess
 import sys
 from queue import Empty, Queue
 from threading import Thread
-from typing import Any, Iterator, Optional
+from typing import Any, BinaryIO, Iterator, Optional
 
 
 class JsonRpcError(Exception):
     """JSON-RPC error."""
 
 
+class RpcShutdownError(JsonRpcError):
+    """Raised in RPC methods if the connection to server is closing."""
+
+
 class RpcMethod:
     """RPC method."""
 
-    def __init__(self, rpc: "Rpc", name: str):
+    def __init__(self, rpc: "BaseRpc", name: str):
         self.rpc = rpc
         self.name = name
 
@@ -44,6 +48,8 @@ class RpcMethod:
         def rpc_future():
             """Wait for the request to receive a result."""
             response = queue.get()
+            if response is None:
+                raise RpcShutdownError(f"no response for {request_id}/{self.name} while rpc is shutting down")
             if "error" in response:
                 raise JsonRpcError(response["error"])
             return response.get("result", None)
@@ -51,40 +57,26 @@ class RpcMethod:
         return rpc_future
 
 
-class Rpc:
-    """RPC client."""
+class BaseRpc:
+    """Base Rpc class which requires 'connect_to_server' and 'disconnect_from_server' methods
+    from subclasses to work concretely."""
 
-    def __init__(
-        self,
-        accounts_dir: Optional[str] = None,
-        rpc_server_path="deltachat-rpc-server",
-        **kwargs,
-    ):
-        """Initialize RPC client.
-
-        The 'kwargs' arguments will be passed to subprocess.Popen().
-        """
-        if accounts_dir:
-            kwargs["env"] = {
-                **kwargs.get("env", os.environ),
-                "DC_ACCOUNTS_PATH": str(accounts_dir),
-            }
-
-        self._kwargs = kwargs
-        self.rpc_server_path = rpc_server_path
-        self.process: subprocess.Popen
+    def __init__(self):
         self.id_iterator: Iterator[int]
         self.event_queues: dict[int, Queue]
         # Map from request ID to a Queue which provides a single result
         self.request_results: dict[int, Queue]
         self.request_queue: Queue[Any]
+        self.server_stdin: BinaryIO
+        self.server_stdout: BinaryIO
         self.closing: bool
+        self.reader_stopping: bool
         self.reader_thread: Thread
         self.writer_thread: Thread
         self.events_thread: Thread
 
     def start(self) -> None:
-        """Start RPC server subprocess and wait for successful initialization.
+        """Connect to the RPC server and wait for successful initialization.
 
         This method blocks until the RPC server responds to an initial
         health-check RPC call (get_system_info).
@@ -92,22 +84,13 @@ class Rpc:
         (e.g., due to an invalid accounts directory),
         a JsonRpcError is raised.
         """
-        popen_kwargs = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE}
-        if sys.version_info >= (3, 11):
-            # Prevent subprocess from capturing SIGINT.
-            popen_kwargs["process_group"] = 0
-        else:
-            # `process_group` is not supported before Python 3.11.
-            popen_kwargs["preexec_fn"] = os.setpgrp  # noqa: PLW1509
-
-        popen_kwargs.update(self._kwargs)
-        self.process = subprocess.Popen(self.rpc_server_path, **popen_kwargs)
-
+        self.server_stdout, self.server_stdin = self.connect_to_server()
         self.id_iterator = itertools.count(start=1)
         self.event_queues = {}
         self.request_results = {}
         self.request_queue = Queue()
         self.closing = False
+        self.reader_stopping = False
         self.reader_thread = Thread(target=self.reader_loop)
         self.reader_thread.start()
         self.writer_thread = Thread(target=self.writer_loop)
@@ -120,24 +103,25 @@ class Rpc:
         try:
             system_info = self.get_system_info()
         except (JsonRpcError, Exception) as e:
-            # The reader_loop already saw EOF on stdout, so the process
-            # has exited and stderr is available.
-            stderr = self.process.stderr.read().decode(errors="replace").strip()
-            if stderr:
-                raise JsonRpcError(f"RPC server failed to start: {stderr}") from e
+            details = self.get_startup_error_details()
+            if details:
+                raise JsonRpcError(f"RPC server failed to start: {details}") from e
             raise JsonRpcError(f"RPC server startup check failed: {e}") from e
         logging.info(
             "RPC server ready. Core version: %s",
             system_info.get("deltachat_core_version", "unknown"),
         )
 
+    def get_startup_error_details(self) -> str:
+        """Return server-side diagnostics for a failed startup."""
+        return ""
+
     def close(self) -> None:
         """Terminate RPC server process and wait until the reader loop finishes."""
         self.closing = True
-        self.stop_io_for_all_accounts()
-        self.events_thread.join()
-        self.process.stdin.close()
+        self.disconnect_from_server()
         self.reader_thread.join()
+        self.events_thread.join()
         self.request_queue.put(None)
         self.writer_thread.join()
 
@@ -151,7 +135,11 @@ class Rpc:
     def reader_loop(self) -> None:
         """Process JSON-RPC responses from the RPC server process output."""
         try:
-            while line := self.process.stdout.readline():
+            while line := self.server_stdout.readline():
+                # not self.closing: the subprocess server still answers the
+                # stop_io_for_all_accounts() that close() sends
+                if self.reader_stopping:
+                    break
                 response = json.loads(line)
                 if "id" in response:
                     response_id = response["id"]
@@ -162,17 +150,17 @@ class Rpc:
             # Log an exception if the reader loop dies.
             logging.exception("Exception in the reader loop")
         finally:
-            # Unblock any pending requests when the server closes stdout.
-            for _request_id, queue in self.request_results.items():
-                queue.put({"error": {"code": -32000, "message": "RPC server closed"}})
+            # terminate pending rpc requests because no responses can arrive anymore
+            for queue in list(self.request_results.values()):
+                queue.put(None)
 
     def writer_loop(self) -> None:
         """Writer loop ensuring only a single thread writes requests."""
         try:
             while request := self.request_queue.get():
                 data = (json.dumps(request) + "\n").encode()
-                self.process.stdin.write(data)
-                self.process.stdin.flush()
+                self.server_stdin.write(data)
+                self.server_stdin.flush()
         except Exception:
             # Log an exception if the writer loop dies.
             logging.exception("Exception in the writer loop")
@@ -195,6 +183,9 @@ class Rpc:
                     queue.put(payload)
                 if self.closing:
                     return
+        except RpcShutdownError:
+            # The server connection went away while waiting for the next batch.
+            return
         except Exception:
             # Log an exception if the event loop dies.
             logging.exception("Exception in the event loop")
@@ -215,3 +206,79 @@ class Rpc:
 
     def __getattr__(self, attr: str):
         return RpcMethod(self, attr)
+
+
+class RpcSubprocess(BaseRpc):
+    """RPC client that runs and connects to a deltachat-rpc-server in a subprocess."""
+
+    def __init__(self, accounts_dir: Optional[str] = None, rpc_server_path="deltachat-rpc-server"):
+        super(RpcSubprocess, self).__init__()
+        self._accounts_dir = accounts_dir
+        self.rpc_server_path = rpc_server_path
+        self.process: subprocess.Popen
+
+    def connect_to_server(self):
+        popen_kwargs = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE}
+        if sys.version_info >= (3, 11):
+            # Prevent subprocess from capturing SIGINT.
+            popen_kwargs["process_group"] = 0
+        else:
+            # `process_group` is not supported before Python 3.11.
+            popen_kwargs["preexec_fn"] = os.setpgrp  # noqa: PLW1509
+
+        if self._accounts_dir:
+            popen_kwargs["env"] = os.environ.copy()
+            popen_kwargs["env"]["DC_ACCOUNTS_PATH"] = str(self._accounts_dir)
+
+        self.process = subprocess.Popen(self.rpc_server_path, **popen_kwargs)
+        return self.process.stdout, self.process.stdin
+
+    def get_startup_error_details(self) -> str:
+        # The reader_loop already saw EOF on stdout, so the process
+        # has exited and stderr is available.
+        return self.process.stderr.read().decode(errors="replace").strip()
+
+    def disconnect_from_server(self):
+        self.stop_io_for_all_accounts()
+        self.server_stdin.close()
+
+
+# backward compatibility
+Rpc = RpcSubprocess
+
+
+class RpcFIFO(BaseRpc):
+    """RPC client connecting to an already running deltachat-rpc-server.
+
+    Only use one client per FIFO pair: a FIFO has no per-client routing,
+    so a second client would steal responses and events.
+    """
+
+    def __init__(self, fn_request_fifo: str, fn_response_fifo: str):
+        super(RpcFIFO, self).__init__()
+        self.fn_request_fifo = fn_request_fifo
+        self.fn_response_fifo = fn_response_fifo
+
+    def connect_to_server(self):
+        server_stdin = open(self.fn_request_fifo, "wb")  # noqa
+        server_stdout = open(self.fn_response_fifo, "rb")  # noqa
+        return server_stdout, server_stdin
+
+    def disconnect_from_server(self):
+        self.server_stdin.close()
+        # a server outliving this client keeps the response FIFO open, so closing
+        # it would deadlock on the buffer lock held by the blocked readline()
+        self.reader_stopping = True
+        self.wakeup_reader()
+        self.reader_thread.join()
+        self.server_stdout.close()
+
+    def wakeup_reader(self) -> None:
+        try:
+            fd = os.open(self.fn_response_fifo, os.O_WRONLY | os.O_NONBLOCK)
+        except OSError:
+            return  # nobody is reading anymore, so there is nothing to wake
+        try:
+            os.write(fd, b"\n")
+        finally:
+            os.close(fd)

deltachat-rpc-client/tests/test_rpc_fifo.py

@@ -0,0 +1,51 @@
+import os
+import platform  # noqa
+import subprocess
+import threading
+
+import pytest
+
+from deltachat_rpc_client import DeltaChat, RpcFIFO
+
+
+@pytest.mark.skipif("platform.system() == 'Windows'")
+def test_rpc_fifo(tmp_path):
+    fn_request_fifo = tmp_path.joinpath("request_fifo")
+    fn_response_fifo = tmp_path.joinpath("response_fifo")
+    os.mkfifo(fn_request_fifo)
+    os.mkfifo(fn_response_fifo)
+    # without DC_ACCOUNTS_PATH the server creates an "accounts" dir in the current
+    # working directory, which during a test run is inside the checkout
+    env = {**os.environ, "DC_ACCOUNTS_PATH": str(tmp_path.joinpath("accounts"))}
+    popen = subprocess.Popen(f"deltachat-rpc-server <{fn_request_fifo} >{fn_response_fifo}", shell=True, env=env)
+
+    rpc = RpcFIFO(fn_response_fifo=fn_response_fifo, fn_request_fifo=fn_request_fifo)
+    with rpc:
+        dc = DeltaChat(rpc)
+        assert dc.rpc.get_system_info()["deltachat_core_version"] is not None
+    popen.wait()
+
+
+@pytest.mark.skipif("platform.system() == 'Windows'")
+def test_rpc_fifo_close_with_persistent_server(tmp_path):
+    """close() must return even if the server keeps the response FIFO open."""
+    fn_request_fifo = tmp_path.joinpath("request_fifo")
+    fn_response_fifo = tmp_path.joinpath("response_fifo")
+    os.mkfifo(fn_request_fifo)
+    os.mkfifo(fn_response_fifo)
+    env = {**os.environ, "DC_ACCOUNTS_PATH": str(tmp_path.joinpath("accounts"))}
+    popen = subprocess.Popen(f"exec deltachat-rpc-server <{fn_request_fifo} >{fn_response_fifo}", shell=True, env=env)
+
+    try:
+        # keep the request FIFO open so the server never sees EOF on its stdin
+        with open(fn_request_fifo, "wb"):
+            rpc = RpcFIFO(fn_response_fifo=fn_response_fifo, fn_request_fifo=fn_request_fifo)
+            rpc.start()
+            assert rpc.get_system_info()["deltachat_core_version"] is not None
+
+            closed = threading.Event()
+            threading.Thread(target=lambda: (rpc.close(), closed.set()), daemon=True).start()
+            assert closed.wait(timeout=30), "RpcFIFO.close() did not return"
+    finally:
+        popen.kill()
+        popen.wait()