fix(dhcp4): make the DHCPv4 client INIT path responsive to stop()

'pytcp stack start' could not be interrupted: a Ctrl-C ran the full
teardown, but the DHCPv4 client worker — parked in the RFC 2131 §4.4.1
startup desync 'time.sleep' (or a multi-second blocking recv in the
retransmission backoff) — missed the bounded 2.0 s join in
'Subsystem.stop()' and was left dangling. Because the worker was a
non-daemon thread, it then kept the interpreter alive AND ran the entire
acquisition (DISCOVER + 5 retransmits, ~2 min against a silent server),
so further Ctrl-Cs did nothing.

Two fixes:

- DHCPv4 client: the desync delay and the post-DECLINE backoff now wait
  on the stop event instead of 'time.sleep' (interruptible; in sync
  'fetch()' the event is never set, so they still behave as a sleep),
  '_do_init_to_bound' aborts before opening a socket / before each
  NAK-restart when stop is set, and '_recv_with_backoff' bails between
  attempts. A daemon-mode stop() during INIT now halts immediately
  without further wire traffic.

- Subsystem: spawn the worker as a daemon thread. The bounded join in
  stop() still gives a graceful exit; the daemon flag is the safety net
  so a worker stuck in a syscall is abandoned at interpreter exit rather
  than wedging the process.

Tests-first. New TestDhcp4ClientStopResponsive pins: abort-before-socket
on stop, recv-backoff immediate bail, interruptible desync wait, and a
real-thread daemon stop() during the desync delay that does not leave
the worker dangling. New test__subsystem__worker_thread_is_daemon pins
the daemon flag. The three initial-delay tests and the decline-backoff
test are updated to assert the interruptible stop-event wait rather than
'time.sleep'.

Reference: RFC 2131 §4.4.1 (startup desynchronisation delay).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ccie18643

packages/pytcp/pytcp/protocols/dhcp4/dhcp4__client.py

@@ -1351,6 +1351,10 @@ def _do_init_to_bound(self) -> Dhcp4Lease | None:
         """
 
         self._initial_delay()
+        # A daemon-mode 'stop()' during the desync delay returns here
+        # without opening a socket or emitting any wire traffic.
+        if self._event__stop_subsystem.is_set():
+            return None
         self._fetch_started_at_monotonic = time.monotonic()
         __debug__ and log(
             "dhcp4",
@@ -1363,6 +1367,8 @@ def _do_init_to_bound(self) -> Dhcp4Lease | None:
             client_socket.connect(("255.255.255.255", 67))
 
             for _ in range(dhcp4__constants.DHCP4__NAK_MAX_RESTARTS + 1):
+                if self._event__stop_subsystem.is_set():
+                    return None
                 outcome = self._discover_request_once(client_socket)
                 if not isinstance(outcome, _NakRestart):
                     if isinstance(outcome, Dhcp4Lease):
@@ -1484,7 +1490,9 @@ def _discover_request_once(self, client_socket: socket) -> Dhcp4Lease | _NakRest
             )
             backoff_s = dhcp4__constants.DHCP4__DECLINE_BACKOFF_MS / 1000.0
             if backoff_s > 0:
-                time.sleep(backoff_s)
+                # Interruptible wait so a daemon-mode 'stop()' during the
+                # post-DECLINE backoff is not wedged for the full window.
+                self._event__stop_subsystem.wait(timeout=backoff_s)
             return _NAK_RESTART
 
         t1_override, t2_override = self._extract_t1_t2_overrides(ack, ack.lease_time)
@@ -1529,6 +1537,12 @@ def _recv_with_backoff(
         jitter_ms = dhcp4__constants.DHCP4__RETRANS_JITTER_MS
 
         for attempt in range(max_attempts):
+            # Bail between attempts so a daemon-mode 'stop()' halts the
+            # retransmission storm promptly instead of running the full
+            # backoff budget (which can span minutes against a silent
+            # server). The event is never set in sync 'fetch()'.
+            if self._event__stop_subsystem.is_set():
+                return None
             jitter_s = random.uniform(-jitter_ms / 1000.0, jitter_ms / 1000.0)
             timeout_s = max(0.001, delay_ms / 1000.0 + jitter_s)
             result = self._recv_within_window(
@@ -1828,7 +1842,11 @@ def _initial_delay(self) -> None:
         min_ms = dhcp4__constants.DHCP4__INIT_DELAY_MIN_MS
         delay_s = random.uniform(min_ms / 1000.0, max_ms / 1000.0)
         __debug__ and log("dhcp4", f"Initial desync delay: {delay_s:.2f}s")
-        time.sleep(delay_s)
+        # Interruptible wait, not 'time.sleep': in daemon mode a 'stop()'
+        # during the desync window must wake the worker immediately so
+        # 'stack.stop()' is not wedged for the full delay. In sync
+        # 'fetch()' the event is never set, so this behaves as a sleep.
+        self._event__stop_subsystem.wait(timeout=delay_s)
 
     def _elapsed_secs(self) -> int:
         """

packages/pytcp/pytcp/runtime/subsystem.py

@@ -109,7 +109,13 @@ def start(self) -> None:
         )
 
         self._event__stop_subsystem.clear()
-        self._thread = threading.Thread(target=self._thread__subsystem)
+        # Daemon worker: 'stop()' joins it with a bounded 2.0 s timeout
+        # for a graceful exit, but a subclass blocked in a syscall (a
+        # DHCPv4 client mid-recv, a ring blocked on the TAP fd) can miss
+        # that window and be left dangling. A daemon thread is then
+        # abandoned at interpreter exit instead of wedging the process —
+        # the safety net behind the bounded join in 'stop()'.
+        self._thread = threading.Thread(target=self._thread__subsystem, daemon=True)
         self._thread.start()
         self._start()
 

packages/pytcp/pytcp/tests/unit/protocols/dhcp4/test__dhcp4__client.py

@@ -31,6 +31,7 @@
 """
 
 import errno
+import threading
 from typing import Any, cast, override
 from unittest import TestCase
 from unittest.mock import MagicMock, create_autospec, patch
@@ -1156,14 +1157,20 @@ def test__dhcp4_client__fetch_initial_delay_uses_default_bounds(self) -> None:
         self._server.enqueue_offer()
         self._server.enqueue_ack()
 
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
         with (
             patch("pytcp.protocols.dhcp4.dhcp4__client.random.uniform", return_value=4.2) as mock_uniform,
+            patch.object(client._event__stop_subsystem, "wait") as mock_wait,
             patch("pytcp.protocols.dhcp4.dhcp4__client.time.sleep") as mock_sleep,
         ):
-            Dhcp4Client(mac_address=_DEFAULT_MAC).fetch()
+            client.fetch()
 
         mock_uniform.assert_any_call(1.0, 10.0)
-        mock_sleep.assert_any_call(4.2)
+        # The desync delay is an interruptible wait on the stop event, not
+        # an uninterruptible 'time.sleep', so a daemon-mode stop() during
+        # the window wakes it immediately.
+        mock_wait.assert_any_call(timeout=4.2)
+        mock_sleep.assert_not_called()
 
     def test__dhcp4_client__fetch_initial_delay_honours_custom_sysctl_bounds(self) -> None:
         """
@@ -1179,11 +1186,12 @@ def test__dhcp4_client__fetch_initial_delay_honours_custom_sysctl_bounds(self) -
         self._server.enqueue_offer()
         self._server.enqueue_ack()
 
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
         with (
             patch("pytcp.protocols.dhcp4.dhcp4__client.random.uniform", return_value=1.0) as mock_uniform,
-            patch("pytcp.protocols.dhcp4.dhcp4__client.time.sleep"),
+            patch.object(client._event__stop_subsystem, "wait"),
         ):
-            Dhcp4Client(mac_address=_DEFAULT_MAC).fetch()
+            client.fetch()
 
         # 500 ms / 1000 = 0.5 s; 2500 ms / 1000 = 2.5 s.
         mock_uniform.assert_any_call(0.5, 2.5)
@@ -1193,24 +1201,153 @@ def test__dhcp4_client__fetch_initial_delay_disabled_when_max_ms_zero(self) -> N
         Ensure the startup desync delay is bypassed entirely when
         'dhcp.init_delay_max_ms' is 0 — the canonical
         disable-for-tests configuration that the fixture base
-        applies by default. 'time.sleep' must not be invoked from
-        the initial-delay path on the happy-path 'fetch()'.
+        applies by default. The interruptible stop-event wait must
+        not be invoked from the initial-delay path on the happy-path
+        'fetch()'.
 
         Reference: PyTCP test infrastructure (no RFC clause).
         """
 
         # Fixture base already sets both bounds to 0; no override
         # needed here. The Phase 1 backoff path uses 'recv__mv(timeout=)'
-        # rather than 'time.sleep', so 'time.sleep' should not be
-        # called at all during a happy-path fetch.
+        # rather than the stop-event wait, so the desync wait should not
+        # be called at all during a happy-path fetch.
         self._server.enqueue_offer()
         self._server.enqueue_ack()
 
-        with patch("pytcp.protocols.dhcp4.dhcp4__client.time.sleep") as mock_sleep:
-            Dhcp4Client(mac_address=_DEFAULT_MAC).fetch()
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
+        with patch.object(client._event__stop_subsystem, "wait") as mock_wait:
+            client.fetch()
+
+        mock_wait.assert_not_called()
+
+
+class TestDhcp4ClientStopResponsive(_Dhcp4ClientFixture):
+    """
+    The daemon-mode stop-responsiveness tests — the INIT acquisition
+    path must observe the stop event promptly so 'stack.stop()' is not
+    wedged by a DHCPv4 client mid-acquisition.
+    """
+
+    def test__dhcp4_client__do_init_to_bound_aborts_before_socket_when_stop_set(self) -> None:
+        """
+        Ensure '_do_init_to_bound' returns None without opening a client
+        socket when the stop event is already set, so a stop() arriving
+        during the startup desync window does not proceed to a wire
+        exchange.
+
+        Reference: PyTCP test infrastructure (no RFC clause).
+        """
+
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
+        client._event__stop_subsystem.set()
+
+        result = client._do_init_to_bound()
+
+        self.assertIsNone(
+            result,
+            msg="A stop-signalled INIT must abandon acquisition and return None.",
+        )
+        self._socket_factory.assert_not_called()
+
+    def test__dhcp4_client__recv_backoff_bails_immediately_when_stop_set(self) -> None:
+        """
+        Ensure '_recv_with_backoff' returns None without issuing any
+        recv or invoking the resend callback when the stop event is
+        already set, so the retransmission storm halts the instant
+        stop() is requested.
+
+        Reference: PyTCP test infrastructure (no RFC clause).
+        """
+
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
+        client._event__stop_subsystem.set()
+        resend = MagicMock()
+
+        result = client._recv_with_backoff(
+            self._sock,
+            expected_type=Dhcp4MessageType.ACK,
+            xid=_PINNED_XID,
+            resend=resend,
+        )
 
+        self.assertIsNone(
+            result,
+            msg="A stop-signalled recv backoff must return None without waiting.",
+        )
+        self._sock.recv__mv.assert_not_called()
+        resend.assert_not_called()
+
+    def test__dhcp4_client__initial_delay_waits_on_interruptible_stop_event(self) -> None:
+        """
+        Ensure the startup desync delay waits on the stop event (so it
+        is interruptible) rather than calling the uninterruptible
+        'time.sleep'.
+
+        Reference: RFC 2131 §4.4.1 (random startup desync delay).
+        """
+
+        self.enterContext(sysctl.override("dhcp.init_delay_min_ms", 1000))
+        self.enterContext(sysctl.override("dhcp.init_delay_max_ms", 10000))
+        self._server.enqueue_offer()
+        self._server.enqueue_ack()
+
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
+        with (
+            patch("pytcp.protocols.dhcp4.dhcp4__client.random.uniform", return_value=3.0),
+            patch.object(client._event__stop_subsystem, "wait") as mock_wait,
+            patch("pytcp.protocols.dhcp4.dhcp4__client.time.sleep") as mock_sleep,
+        ):
+            client.fetch()
+
+        mock_wait.assert_any_call(timeout=3.0)
         mock_sleep.assert_not_called()
 
+    def test__dhcp4_client__daemon_stop_during_init_delay_does_not_dangle(self) -> None:
+        """
+        Ensure a daemon-mode 'stop()' issued while the worker is parked
+        in the startup desync delay terminates the worker thread instead
+        of leaving it dangling for the full delay (the wedge that made
+        'pytcp stack start' unstoppable).
+
+        Reference: PyTCP test infrastructure (no RFC clause).
+        """
+
+        self.enterContext(sysctl.override("dhcp.init_delay_min_ms", 5000))
+        self.enterContext(sysctl.override("dhcp.init_delay_max_ms", 5000))
+        self.enterContext(patch("pytcp.runtime.subsystem.log"))
+
+        reached_delay = threading.Event()
+
+        def _uniform(*args: float) -> float:
+            del args
+            reached_delay.set()
+            return 5.0
+
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC)
+        # Captured the instant stop() returns (after its bounded 2.0 s
+        # join) — BEFORE the cleanup join below, which would otherwise
+        # mask a wedged worker by waiting out the full desync delay.
+        alive_after_stop = True
+        with patch("pytcp.protocols.dhcp4.dhcp4__client.random.uniform", side_effect=_uniform):
+            client.start()
+            try:
+                self.assertTrue(
+                    reached_delay.wait(timeout=5.0),
+                    msg="Precondition: the worker must reach the desync delay.",
+                )
+                client.stop()
+                assert client._thread is not None
+                alive_after_stop = client._thread.is_alive()
+            finally:
+                if client._thread is not None:
+                    client._thread.join(timeout=10.0)
+
+        self.assertFalse(
+            alive_after_stop,
+            msg="stop() during the desync delay must terminate the worker promptly.",
+        )
+
 
 class TestDhcp4ClientFetchArpDad(_Dhcp4ClientFixture):
     """
@@ -1415,15 +1552,18 @@ def test__dhcp4_client__fetch_decline_path_honours_decline_backoff_sleep(self) -
 
         acd = _acd_mock(probe_results=False, conflict_mac=MacAddress("02:00:00:00:00:99"))
 
-        with patch("pytcp.protocols.dhcp4.dhcp4__client.time.sleep") as mock_sleep:
-            Dhcp4Client(
-                mac_address=_DEFAULT_MAC,
-                acd=acd,
-            ).fetch()
+        client = Dhcp4Client(mac_address=_DEFAULT_MAC, acd=acd)
+        with (
+            patch.object(client._event__stop_subsystem, "wait") as mock_wait,
+            patch("pytcp.protocols.dhcp4.dhcp4__client.time.sleep") as mock_sleep,
+        ):
+            client.fetch()
 
-        # Initial-delay sleep is disabled by the fixture; only the
-        # decline-backoff sleep should fire.
-        mock_sleep.assert_called_once_with(5.0)
+        # Initial-delay wait is disabled by the fixture; only the
+        # decline-backoff wait should fire — and as an interruptible
+        # stop-event wait, not an uninterruptible 'time.sleep'.
+        mock_wait.assert_called_once_with(timeout=5.0)
+        mock_sleep.assert_not_called()
 
 
 class TestDhcp4ClientFetchRfc4361Cid(_Dhcp4ClientFixture):

packages/pytcp/pytcp/tests/unit/runtime/test__runtime__subsystem.py

@@ -329,6 +329,33 @@ def test__subsystem__start_runs_thread_and_fires_hooks(self) -> None:
         finally:
             subsystem.stop()
 
+    def test__subsystem__worker_thread_is_daemon(self) -> None:
+        """
+        Ensure 'start()' spawns the worker as a daemon thread so a
+        subsystem blocked in a syscall (e.g. a DHCPv4 client mid-recv)
+        can never wedge interpreter exit after the bounded join in
+        'stop()' times out and leaves it dangling.
+
+        Reference: PyTCP test infrastructure (no RFC clause).
+        """
+
+        subsystem = _TestSubsystem()
+
+        try:
+            subsystem.start()
+
+            self.assertTrue(
+                subsystem._loop_event.wait(timeout=2.0),
+                msg="Precondition: the worker thread must be running.",
+            )
+            assert subsystem._thread is not None
+            self.assertTrue(
+                subsystem._thread.daemon,
+                msg="Subsystem.start() must spawn the worker as a daemon thread.",
+            )
+        finally:
+            subsystem.stop()
+
     def test__subsystem__stop_signals_event_and_fires_hook(self) -> None:
         """
         Ensure 'stop()' sets the stop event (terminating the loop) and