Fix pystemd.run waits and pystemd.futures on systemd 259 / Python 3.14

CHANGES.md

@@ -6,6 +6,20 @@ the one you find pypi, but its the one in development.
 
 dev
 =====
+* Fix `pystemd.run(wait=True)` / `wait_for_activation=True` hanging forever on hosts where no other client is subscribed to systemd signals
+* Fix `pystemd.run` missing unit state changes when several D-Bus signals arrive at once
+* Fix `pystemd.futures` on Python 3.14+, where the default multiprocessing start method is no longer fork
+* Fix `pystemd.futures` ignoring `User=` on systemd >= 255
+* Fix `pystemd.futures.TransientUnitPoolExecutor` leaking its unit when the parent process is killed
+* Add `PrivatePIDs`, `ProtectHostnameEx` and `SystemCallErrorNumber` to systemd1 KNOWN_UNIT_SIGNATURES
+* Fix `pystemd.run(wait_for_activation=True)` hanging forever when the unit fails to start
+* Add `FileDescriptorStoreMax` and `FileDescriptorStorePreserve` to systemd1 KNOWN_UNIT_SIGNATURES
+* Add `fds=` to `pystemd.daemon.notify` (sd_pid_notify_with_fds) and add `pystemd.daemon.listen_fds_with_names` (file descriptor store / named socket activation)
+* Add `pystemd.daemon.store_fd` and `pystemd.daemon.store_fds` helpers for the file descriptor store
+* Split `pystemd.utils` into a package (`pystemd.utils.{x2,unit,typing}`); existing imports keep working
+* Add `pystemd.utils.ns.join_namespaces` (atomic pidfd `setns`, with a per-namespace fallback); `pystemd.futures` uses it, which also fixes leaking one fd per namespace in every worker
+* Add `delegate=` to `pystemd.run`, a shortcut for `extra={"Delegate": ...}`
+* Add `OOMPolicy`, `OOMScoreAdjust` and the `ManagedOOM*` properties to systemd1 KNOWN_UNIT_SIGNATURES
 
 0.15.3
 =====

pystemd/daemon.pyi

@@ -5,7 +5,12 @@
 # the root directory of this source tree.
 #
 
-from typing import Union
+from typing import Iterable, Mapping, Optional, Protocol, Union
+
+class _HasFileno(Protocol):
+    def fileno(self) -> int: ...
+
+_Fds = Union[int, _HasFileno, Iterable[Union[int, _HasFileno]]]
 
 LISTEN_FDS_START: int
 
@@ -14,7 +19,19 @@ class PystemdDaemonError(Exception):
 
 def listen_fds(unset_environment: bool) -> int: ...
 def notify(
-    unset_environment: bool, *args: Union[str, bytes], **kwargs: Union[int, str]
-) -> None: ...
+    unset_environment: bool,
+    *args: Union[str, bytes],
+    fds: Optional[_Fds] = None,
+    **kwargs: Union[int, str],
+) -> int: ...
+def listen_fds_with_names(unset_environment: bool = ...) -> dict[str, list[int]]: ...
+def store_fd(
+    name: str, fds: _Fds, unset_environment: bool = ..., **kwargs: Union[int, str]
+) -> int: ...
+def store_fds(
+    fds_by_name: Mapping[str, _Fds],
+    unset_environment: bool = ...,
+    **kwargs: Union[int, str],
+) -> int: ...
 def booted() -> bool: ...
 def watchdog_enabled(unset_environment: bool) -> int: ...

pystemd/daemon.pyx

@@ -9,6 +9,7 @@
 
 
 from libc.stdint cimport uint64_t
+from libc.stdlib cimport free, malloc
 
 cimport pystemd.dbusc as dbusc
 
@@ -24,7 +25,39 @@ def listen_fds(int unset_environment=0):
   return dbusc.sd_listen_fds(unset_environment)
 
 
-def notify(int unset_environment, *states, **kwstates):
+def listen_fds_with_names(int unset_environment=0):
+  """
+  interface to systemd sd_listen_fds_with_names: the file descriptors passed
+  to us (socket activation, or the file descriptor store handing back what we
+  stored with FDSTORE=1), as a dict of {name: [fd, ...]}. fds start at
+  LISTEN_FDS_START; names come from FileDescriptorName= / FDNAME=.
+
+  Names are not unique: a socket unit with several ListenStream= passes all
+  its fds under the same name, and FDNAME= can be reused; hence the lists.
+
+  usage:
+    * `fd = pystemd.daemon.listen_fds_with_names()["db"][0]`
+    * `for fd in pystemd.daemon.listen_fds_with_names().get("web.socket", []): ...`
+  """
+  cdef char **names = NULL
+  cdef int n = dbusc.sd_listen_fds_with_names(unset_environment, &names)
+  if n < 0:
+    raise PystemdDaemonError(f"sd_listen_fds_with_names failed: {n}")
+
+  fds = {}
+  try:
+    for i in range(n):
+      name = names[i].decode() if names != NULL else "unknown"
+      fds.setdefault(name, []).append(LISTEN_FDS_START + i)
+    return fds
+  finally:
+    if names != NULL:
+      for i in range(n):
+        free(names[i])
+      free(names)
+
+
+def notify(int unset_environment, *states, fds=None, **kwstates):
   """
   interface to systemd sd_notify,
 
@@ -34,6 +67,11 @@ def notify(int unset_environment, *states, **kwstates):
     states/kwstates = array with states statements that are 'IDENTIFIER=VALUE'.
       e.g: 'READY=1' or keywords states like `ready=1`
       (This will be clear in usage).
+    fds = optional file descriptor, or list of them (ints, or objects with a
+      fileno() method), to send along with the message (sd_pid_notify_with_fds). Mostly
+      for the file descriptor store: with `fdstore=1` systemd keeps them while
+      the service restarts and hands them back on the next start (see
+      listen_fds_with_names); the unit needs FileDescriptorStoreMax= > 0.
 
   usage:
     * `pystemd.daemon.notify(1, ready=1)`: this will signal `READY=1` to systemd
@@ -44,18 +82,114 @@ def notify(int unset_environment, *states, **kwstates):
     *  `pystemd.daemon.notify(False, ready=1, status='gime gime gime')`: will
       signal systemd that the app is ready and also set the status to
       'gime gime gime'. This command does not clean the notify environment.
+    *  `pystemd.daemon.notify(False, fds=[memfd], fdstore=1, fdname="db")`:
+      ask systemd to keep `memfd` for us, under the name "db".
 
     For info on sd_notify, check https://github.com/systemd/systemd/blob/\
 master/src/systemd/sd-daemon.h#L173-L232
 
   """
+  state = _join_states(states, kwstates)
+  if fds is None:
+    return dbusc.sd_notify(unset_environment, state)
+
+  fd_list = _fd_list(fds)
+  if not fd_list:
+    return dbusc.sd_notify(unset_environment, state)
+
+  cdef unsigned n_fds = len(fd_list)
+  cdef int *c_fds = <int *> malloc(n_fds * sizeof(int))
+  if c_fds == NULL:
+    raise MemoryError()
+
+  try:
+    for i, fd in enumerate(fd_list):
+      c_fds[i] = fd
+    return dbusc.sd_pid_notify_with_fds(0, unset_environment, state, c_fds, n_fds)
+  finally:
+    free(c_fds)
+
+
+def store_fd(name, fds, unset_environment=False, **kwstates):
+  """
+  hand `fds` (a single fd or a list; ints or objects with fileno()) to
+  systemd's file descriptor store under `name`: systemd keeps them while the
+  service restarts, and listen_fds_with_names()[name] gives them back on the
+  next start. The unit needs FileDescriptorStoreMax= > 0.
+
+  Same as notify(unset_environment, fds=fds, fdstore=1, fdname=name, **kwstates).
+
+  usage:
+    * `pystemd.daemon.store_fd("db", memfd)`
+    * `pystemd.daemon.store_fd("listener", sock, ready=1)`
+  """
+  _check_fdname(name)
+  return notify(unset_environment, fds=fds, fdstore=1, fdname=name, **kwstates)
+
+
+def store_fds(fds_by_name, unset_environment=False, **kwstates):
+  """
+  store_fd() for several names at once: `fds_by_name` is {name: fds}, where
+  fds is a single fd or a list. One FDNAME= covers a whole notify message,
+  so this sends one message per name (not atomically). `unset_environment`
+  and the extra states (e.g. ready=1) only go with the last message.
+  Returns the lowest notify() result (negative errno if any send failed).
+
+  usage:
+    * `pystemd.daemon.store_fds({"listener": sock, "log": memfd}, ready=1)`
+    * `pystemd.daemon.store_fds({...}, True)` or `store_fds({...}, unset_environment=True)`
+  """
+  items = list(fds_by_name.items())
+  for name, _ in items:
+    _check_fdname(name)
+  if not items:
+    return notify(unset_environment, **kwstates) if kwstates else 0
+
+  results = [store_fd(name, fds) for name, fds in items[:-1]]
+  name, fds = items[-1]
+  results.append(store_fd(name, fds, unset_environment=unset_environment, **kwstates))
+  return min(results)
+
+
+def _check_fdname(name):
+  # systemd ignores invalid names *silently* (the fds are just not stored),
+  # so fail here instead: LISTEN_FDNAMES is ':'-separated, printable ASCII.
+  if (
+    not isinstance(name, str) or not 0 < len(name) <= 255
+    or ":" in name or not all(32 <= ord(c) < 127 for c in name)
+  ):
+    raise ValueError(f"invalid fd name {name!r}: 1-255 printable ASCII characters, no ':'")
+
+
+def _fd_list(fds):
+  """
+  normalize notify()'s `fds`: a single fd (int, or object with fileno()) or an
+  iterable of them, to a list of ints.
+  """
+  if isinstance(fds, (str, bytes, bytearray)):
+    # bytes would iterate as ints, silently sending the wrong fds
+    raise TypeError(f"fds must be a file descriptor or a list of them, not {type(fds).__name__}")
+  if isinstance(fds, int) or hasattr(fds, "fileno"):
+    fds = [fds]
+
+  fd_list = []
+  for fd in fds:
+    if isinstance(fd, int):
+      fd_list.append(fd)
+    elif hasattr(fd, "fileno"):
+      fd_list.append(fd.fileno())
+    else:
+      raise TypeError(f"not a file descriptor: {fd!r} (expected an int or an object with fileno())")
+  return fd_list
+
+
+def _join_states(states, kwstates):
   pystates = [x2char_star(s) for s in states]
   pystates.extend(
     b'='.join([x2char_star(k.upper()), x2char_star(v, convert_all=True)])
     for k, v in kwstates.items()
   )
-  state = b'\n'.join(pystates)
-  return dbusc.sd_notify(unset_environment, state)
+  return b'\n'.join(pystates)
 
 
 def booted():

pystemd/dbusc.pxd

@@ -7,6 +7,7 @@
 # the root directory of this source tree.
 #
 
+from posix.types cimport pid_t
 from libc.stdint cimport (
     int16_t,
     int32_t,
@@ -37,6 +38,9 @@ cdef extern from "systemd/sd-daemon.h":
   int SD_LISTEN_FDS_START
   int sd_listen_fds(int unset_environment)
   int sd_notify(int unset_environment, const char *state)
+  int sd_pid_notify_with_fds(pid_t pid, int unset_environment, const char *state,
+                             const int *fds, unsigned n_fds)
+  int sd_listen_fds_with_names(int unset_environment, char ***names)
   int sd_booted()
   int sd_watchdog_enabled(int unset_environment, uint64_t *usec)
 

pystemd/futures.py

@@ -3,17 +3,15 @@
 import os
 from concurrent.futures import ProcessPoolExecutor
 from functools import cached_property
-from multiprocessing import Process
-from multiprocessing.context import BaseContext
-from pathlib import Path
+from multiprocessing.context import BaseContext, ForkProcess
 from typing import Any, Sequence, cast
 
 import psutil
 
-import pystemd.cutils
 import pystemd.run
 import pystemd.utils
 from pystemd.dbuslib import DBus
+from pystemd.utils.ns import join_namespaces
 
 
 class TransientUnitContext(BaseContext):
@@ -34,10 +32,12 @@ def __init__(
         self.properties = properties
         self.user_mode = user_mode
         self.unit_name = unit_name
+        # The unit's main process lives exactly as long as we do, so the unit
+        # is not leaked if we die without reaching stop_unit() (e.g. SIGTERM).
         self.main_process_cmd = main_process or [
             "/bin/bash",
             "-c",
-            "exec sleep infinity",
+            f"exec tail --pid={os.getpid()} -f /dev/null",
         ]
         super().__init__()
 
@@ -49,6 +49,11 @@ def start_unit(self) -> pystemd.systemd1.Unit:
             name=self.unit_name,
             user_mode=self.user_mode,
             wait_for_activation=True,
+            # "exec": only report the unit running once the main process has
+            # exec'd, i.e. once its user and namespaces are final. With
+            # "simple", enter_unit could read the uid of the not-yet-dropped
+            # executor (systemd >= 255) and keep running as root.
+            service_type="exec",
             extra={
                 **self.properties,
                 "Delegate": True,
@@ -72,7 +77,7 @@ def enter_unit(unit):
     after fork, but before code to run. It basically do:
 
         1) Attach itself to the main transient unit (making it part of the cgroup)
-        2) Change its namespace too all the namespaces of the main transient unit, this is best effort
+        2) Join all the namespaces of the main transient unit (see pystemd.utils.ns.join_namespaces)
         3) Change group and user to the group and user of the main transient unit
     """
     # get the unit information
@@ -84,18 +89,20 @@ def enter_unit(unit):
 
     unit.Service.AttachProcesses("/", [os.getpid()])
 
-    # attach ourselves to the namespace
-    for ns in Path(f"/proc/{main_pid}/ns").iterdir():
-        pystemd.cutils.setns(os.open(ns, os.O_RDONLY), 0)
+    # enter the unit's namespaces (PrivateNetwork=, PrivateTmp=, ...)
+    join_namespaces(main_pid)
 
     # change the user
     os.setgid(gid)
     os.setuid(uid)
 
 
-class _ProcessWithPreRun(Process):
+class _ProcessWithPreRun(ForkProcess):
     """
-    Parent class that its just a subprocess, but it execute a pre_run hook
+    Parent class that its just a subprocess, but it execute a pre_run hook.
+
+    Always forks (enter_unit relies on it), even on Python 3.14+ where the
+    default start method on Linux is no longer fork.
     """
 
     def __init__(self, *args, **kwargs) -> None: