nixos/test-driver: keep nspawn notify socket drained after boot (#533429)

nixos/lib/test-driver/src/test_driver/machine/__init__.py

@@ -1450,6 +1450,7 @@ class NspawnMachine(BaseMachine):
 
     machine_sock_path: Path
     machine_sock: socket.socket | None
+    notify_thread: threading.Thread | None
 
     @staticmethod
     def machine_name_from_start_command(start_command: str) -> str:
@@ -1480,6 +1481,12 @@ class NspawnMachine(BaseMachine):
 
         self.start_command = start_command
         self.process = None
+        self.notify_thread = None
+        # State maintained by the notify-socket drainer thread (see
+        # `_drain_notify_socket`). Guarded by `_notify_lock`.
+        self._notify_lock = threading.Lock()
+        self._notify_ready = False
+        self._notify_leader_pid: int | None = None
 
         self.machine_sock_path = self.tmp_dir / f"{self.name}-nspawn.sock"
 
@@ -1514,43 +1521,76 @@ class NspawnMachine(BaseMachine):
     def is_up(self) -> bool:
         return self.process is not None
 
-    def _poll_socket(self) -> tuple[bool, int | None]:
-        """Non-blocking check of container status via socket.
-        Returns (is_ready, leader_pid).
+    def _drain_notify_socket(self) -> None:
+        """Continuously drain the container's `sd_notify` socket (NOTIFY_SOCKET)
+        for the whole lifetime of the container, recording readiness and the
+        leader PID as they arrive.
+
+        Draining must not stop after boot: the container's PID 1 re-sends
+        `READY=1` on every `systemctl daemon-reexec` (the same Manager.Reexecute
+        that switch-to-configuration issues on a systemd change). If nothing
+        reads the socket, its receive buffer fills and PID 1 blocks in
+        `sendmsg()` to NOTIFY_SOCKET while re-executing -- it never finishes
+        re-initializing, and every later `systemctl` call inside the container
+        hangs or fails with `Transport endpoint is not connected`.
         """
         assert self.machine_sock is not None
-        ready = False
-        leader_pid = None
-        try:
-            data, _ = self.machine_sock.recvfrom(4096)
-            msg = data.decode()
-            for line in msg.splitlines():
+        sock = self.machine_sock
+        proc = self.process
+        assert proc is not None
+        # Bound the thread to the container's lifetime: on
+        # `wait_for_shutdown()` only non-None `proc.poll()` ends the loop.
+        # On exit of PID 1, any datagrams still queued are stale, so drop them.
+        while proc.poll() is None:
+            try:
+                # Block (with a timeout so we notice the container exiting)
+                # rather than busy-poll; we just need to keep the buffer empty.
+                sock.settimeout(0.5)
+                data, _ = sock.recvfrom(4096)
+            except (TimeoutError, BlockingIOError):
+                continue
+            except OSError:
+                break
+            ready = False
+            leader_pid = None
+            for line in data.decode(errors="replace").splitlines():
                 if line == "READY=1":
                     ready = True
                 if line.startswith("X_NSPAWN_LEADER_PID="):
                     leader_pid = int(line.split("=")[1])
-        except OSError:
-            pass
-        return ready, leader_pid
+            if ready or leader_pid is not None:
+                with self._notify_lock:
+                    if ready:
+                        self._notify_ready = True
+                    if leader_pid is not None:
+                        self._notify_leader_pid = leader_pid
 
     @cached_property
     def get_systemd_process(self) -> int:
-        """Block until startup is complete and return the PID of the container's systemd process."""
-        assert self.process is not None
+        """Block until startup is complete and return the PID of the container's systemd process.
 
-        container_pid: int | None = None
-        is_ready = False
+        Readiness and the leader PID are reported over NOTIFY_SOCKET, which is
+        drained by `_drain_notify_socket` (started in `start()`); we just wait
+        for that thread to record both.
+        """
+        assert self.process is not None
 
         start_time = time.monotonic()
         last_warning = start_time
         delay = 0.01
         max_delay = 0.5
 
-        while not is_ready or container_pid is None:
-            # Poll the socket until we have the container leader PID
+        # Poll the socket until we have the container leader PID
+        while True:
             if self.process.poll() is not None:
                 raise MachineError("systemd-nspawn process exited unexpectedly")
 
+            with self._notify_lock:
+                is_ready = self._notify_ready
+                container_pid = self._notify_leader_pid
+            if is_ready and container_pid is not None:
+                return container_pid
+
             # Print periodic warnings every 10s so the user knows we aren't deadlocked
             now = time.monotonic()
             if now - last_warning > 10.0:
@@ -1559,18 +1599,8 @@ class NspawnMachine(BaseMachine):
                 )
                 last_warning = now
 
-            # Poll and update our local tracking variables
-            ready_now, pid_now = self._poll_socket()
-            if ready_now:
-                is_ready = True
-            if pid_now:
-                container_pid = pid_now
-
-            if not (is_ready and container_pid):
-                time.sleep(delay)
-                delay = min(delay * 2, max_delay)
-
-        return container_pid
+            time.sleep(delay)
+            delay = min(delay * 2, max_delay)
 
     def _execute(
         self,
@@ -1684,7 +1714,6 @@ class NspawnMachine(BaseMachine):
 
         self.machine_sock = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_DGRAM)
         self.machine_sock.bind(str(self.machine_sock_path))
-        self.machine_sock.setblocking(False)
 
         self.process = subprocess.Popen(
             [self.start_command],
@@ -1700,6 +1729,13 @@ class NspawnMachine(BaseMachine):
 
         self.log(f"systemd-nspawn running (pid {self.process.pid})")
 
+        # Keep the notify socket drained for the container's whole lifetime, so
+        # PID 1 never blocks re-sending `READY=1` on `daemon-reexec`.
+        self.notify_thread = threading.Thread(
+            target=self._drain_notify_socket, daemon=True
+        )
+        self.notify_thread.start()
+
         journal_thread = threading.Thread(target=self._stream_journal, daemon=True)
         journal_thread.start()
 

nixos/tests/all-tests.nix

@@ -152,6 +152,7 @@ in
     ssh-backdoor = runTestOn [ "x86_64-linux" ] ./nixos-test-driver/ssh-backdoor.nix;
     console-log = runTest ./nixos-test-driver/console-log.nix;
     containers = runTest ./nixos-test-driver/containers.nix;
+    nspawn-daemon-reexec-dbus = runTest ./nspawn-daemon-reexec-dbus.nix;
     skip-typecheck = runTest ./nixos-test-driver/skip-typecheck.nix;
     options-doc-regression = import ./nixos-test-driver/options-doc-regression.nix { inherit pkgs; };
     driver-timeout =

nixos/tests/nspawn-daemon-reexec-dbus.nix

@@ -0,0 +1,79 @@
+# Regression test for an nspawn-only systemd re-exec failure that broke D-Bus.
+#
+# Demonstrates `systemctl show` keeps working on `daemon-reexec`.
+#
+# Trigger: `systemctl daemon-reexec` issues a D-Bus `Manager.Reexecute`, like
+# `switch-to-configuration-ng` on a systemd package change.
+#
+# Root cause: inside nspawn test containers PID 1 re-send `READY=1` on the
+# `NOTIFY_SOCKET` on re-exec. The test driver stopped draining that socket
+# after boot, so until drained its receive buffer filled and `systemctl` hung /
+# errored `Failed to connect to bus: Transport endpoint is not connected`.
+{ ... }:
+{
+  name = "nspawn-daemon-reexec-dbus";
+
+  # `containers.<name>` => systemd-nspawn machine (vs `nodes.<name>` => QEMU).
+  # An empty container boots full systemd + D-Bus, which is all we need.
+  containers.machine = { };
+
+  testScript = # python
+    ''
+      import re
+
+      BUS_BROKEN = re.compile(
+          r"Transport endpoint is not connected|Failed to connect to bus"
+      )
+
+      # Without the fix the notify socket's receive buffer fills after 10
+      # undrained `READY=1` resends, so PID 1 blocks then.
+      REEXECS = 10
+
+
+      def bus_broken():
+          """Whether the in-container D-Bus is unusable. A broken bus prints a
+          transport error or hangs until `timeout` kills it (status 124); both
+          count as broken."""
+          status, out = machine.execute(
+              "timeout 10 systemctl show -p ActiveState --value "
+              "multi-user.target 2>&1",
+              check_return=False,
+              timeout=20,
+          )
+          return status != 0 or bool(BUS_BROKEN.search(out)), status, out
+
+
+      machine.start()
+      machine.wait_for_unit("multi-user.target", timeout=120)
+
+      # Pre-reexec sanity: the bus works and shows no break.
+      broken, status, out = bus_broken()
+      assert not broken, (
+          f"bus already broken before any reexec: status={status} out={out!r}"
+      )
+      machine.log(f"pre-reexec sanity OK: {out.strip()!r}")
+
+      broke_at = None
+      for i in range(1, REEXECS + 1):
+          # The same D-Bus Manager.Reexecute that switch-to-configuration issues
+          # on a systemd change.
+          machine.execute(
+              "timeout 30 systemctl daemon-reexec",
+              check_return=False,
+              timeout=45,
+          )
+          broken, status, out = bus_broken()
+          machine.log(f"[reexec {i}] status={status} out={out.strip()!r}")
+          if broken:
+              broke_at = i
+              break
+
+      assert broke_at is None, (
+          f"nspawn D-Bus broke after daemon-reexec #{broke_at} of {REEXECS} "
+          "(systemctl hung or returned a bus transport error). The re-exec'd "
+          "PID 1 never finished re-initialising -- the test driver stopped "
+          "draining the notify socket, so PID 1's READY=1 resend blocked. "
+          "Never observed on QEMU."
+      )
+    '';
+}