fix(dtls): make reader-thread death visible and fail fast

The reader loop exited silently on any socket error, leaving conn/sock
set so the session still looked open. Every later get()/post()/ping()
then waited out its full request timeout on a session nobody was
reading, raising SessionTimeoutError on repeat, forever.

This started biting in v0.1.3 (d677c72), which moved to connected UDP
sockets: a connected socket surfaces ICMP errors on recv, so one
ECONNREFUSED from a rebooting appliance now killed the reader.

- Advisory ICMP errnos (ECONNREFUSED/EHOSTUNREACH/...) no longer kill
  the reader; the next datagram usually works.
- Real reader exits log at WARNING; close()-driven exits stay quiet.
- A _reader_running Event lets get/post/ping/subscribe/refresh_observes
  fail fast via _check_live() with SessionClosedError instead of waiting
  out a timeout. Callers that never start a reader are unaffected.

Refs QuiteYellow/SmartThings-Local#37

smartthings_local/protocol/dtls_session.py

@@ -18,6 +18,7 @@ Reader thread owns the UDP socket. Callers issue get()/post() and block
 on a per-token Event the reader signals. OBSERVE notifications are
 delivered via the on_notification callback.
 """
+import errno
 import os
 import socket
 import threading
@@ -72,6 +73,21 @@ _BLOCK_ACK_TIMEOUT  = 4.0
 # once the ceiling is measured empirically.
 _DEFAULT_RATE_LIMIT_RPS = 5.0
 
+# ICMP errors a connected UDP socket surfaces on the next recv. On these
+# appliances they show up while the device is rebooting, while it holds an
+# orphaned association, or across a router blip, and the next datagram
+# usually works. UDP delivery was never guaranteed, so treat them as
+# advisory and keep reading. Unconnected sockets never see any of this,
+# which is why the reader survived them before the connected-socket change
+# in d677c72 (v0.1.3).
+_ADVISORY_ERRNOS = frozenset(
+    value for value in (
+        getattr(errno, name, None)
+        for name in ('ECONNREFUSED', 'EHOSTUNREACH', 'ENETUNREACH',
+                     'EHOSTDOWN', 'ENETDOWN')
+    ) if value is not None
+)
+
 class DtlsCoapSession:
     """Single sustained DTLS-CoAP session.
 
@@ -168,6 +184,10 @@ class DtlsCoapSession:
 
         self._stop = threading.Event()
         self._reader_thread = None
+        # Set while the reader owns the socket. Cleared when it exits for
+        # any reason, so callers fail fast through _check_live() instead of
+        # waiting out a request timeout against a session nobody is reading.
+        self._reader_running = threading.Event()
         self._last_send_ts = 0.0
 
     def pace(self) -> None:
@@ -253,11 +273,27 @@ class DtlsCoapSession:
         """Spawn the reader thread. Must be called after connect()."""
         if self.sock is None:
             raise RuntimeError("connect() before start_reader()")
+        self._reader_running.set()
         t = threading.Thread(target=self._reader_loop,
                              daemon=True, name='dtls-reader')
         t.start()
         self._reader_thread = t
 
+    def _check_live(self):
+        """Raise if the session cannot carry a request. A dead reader is
+        as fatal as a closed connection: the socket may still accept
+        sends, but no response will ever be dispatched, so waiting out the
+        request timeout only delays the inevitable SessionClosedError.
+
+        Callers that never start a reader (config-flow style) keep the old
+        behaviour — only the conn check applies while _reader_thread is
+        None."""
+        if self.conn is None:
+            raise SessionClosedError()
+        if self._reader_thread is not None and \
+                not self._reader_running.is_set():
+            raise SessionClosedError()
+
     def join(self):
         """Block until the reader thread exits (i.e. socket dies)."""
         if self._reader_thread is not None:
@@ -374,7 +410,23 @@ class DtlsCoapSession:
                     d = sock.recv(65535)
                 except socket.timeout:
                     continue
-                except (OSError, ValueError):
+                except OSError as e:
+                    if self._stop.is_set():
+                        return          # close() got here first
+                    if e.errno in _ADVISORY_ERRNOS:
+                        logger.debug("reader: advisory %s from %s, continuing",
+                                     errno.errorcode.get(e.errno, e.errno),
+                                     self.host)
+                        continue
+                    logger.warning("reader exiting: socket error %s from %s",
+                                   errno.errorcode.get(e.errno, e.errno),
+                                   self.host)
+                    return
+                except ValueError:
+                    # recv on a socket closed underneath the reader.
+                    if not self._stop.is_set():
+                        logger.warning("reader exiting: socket closed "
+                                       "underneath it")
                     return
                 if not d:
                     continue
@@ -419,6 +471,8 @@ class DtlsCoapSession:
                 if exit_reader:
                     return
         finally:
+            # Reader no longer owns the socket — callers must fail fast.
+            self._reader_running.clear()
             # Make sure pending waiters don't hang if the reader dies.
             for tok, (ev, container) in list(self._pending.items()):
                 container.setdefault('err', SessionClosedError())
@@ -490,8 +544,7 @@ class DtlsCoapSession:
         response — Samsung's server keys per-transfer state on the
         token, and dropping a fresh token on block 1+ silently drops
         the request."""
-        if self.conn is None:
+        self._check_live()
-            raise SessionClosedError()
         tok = self._next_tok()
         blob = b''
         num = 0
@@ -566,8 +619,7 @@ class DtlsCoapSession:
     def post(self, path_segs, body_cbor, timeout=8.0):
         """Single-frame POST with a CBOR-encoded body. Returns
         (code, payload_bytes). body_cbor must already be encoded."""
-        if self.conn is None:
+        self._check_live()
-            raise SessionClosedError()
         tok = self._next_tok()
         mid = self._next_mid()
         opts = [(URI_PATH, s.encode()) for s in path_segs]
@@ -600,8 +652,7 @@ class DtlsCoapSession:
         Real half-open-session detection lives in PollScheduler's
         `last_success_ts`, surfaced through KeepaliveTask's
         `liveness_fn`."""
-        if self.conn is None:
+        self._check_live()
-            raise SessionClosedError()
         mid = self._next_mid()
         self._send_dgram(build_coap(TYPE_CON, 0, mid, b'', []))
         return mid
@@ -618,8 +669,7 @@ class DtlsCoapSession:
         tokens via subscribe. Brief race window where a notify on the
         old token gets dropped as 'stale' — acceptable for a 6h-scale
         safety net."""
-        if self.conn is None:
+        self._check_live()
-            raise SessionClosedError()
         for tok, href in list(self._observe_tokens.items()):
             segs = [s for s in href.split('/') if s]
             try:
@@ -642,8 +692,7 @@ class DtlsCoapSession:
 
         Returns the token used (in case the caller wants to deregister
         later)."""
-        if self.conn is None:
+        self._check_live()
-            raise SessionClosedError()
         tok = self._next_observe_tok()
         href = '/' + '/'.join(path_segs)
         # Register the token BEFORE sending — otherwise the device

tests/test_dtls_session_reader_death.py

@@ -0,0 +1,173 @@
+"""Reader-thread death visibility (QuiteYellow/SmartThings-Local#37).
+
+A connected UDP socket surfaces ICMP errors on recv; before this the
+reader exited silently on the first one and every later request waited
+out its full timeout against a session nobody was reading. These tests
+pin the three behaviours that fixed it: advisory ICMP errnos keep the
+reader alive, a real socket error exits with a WARNING and clears
+_reader_running, and callers then fail fast with SessionClosedError.
+"""
+import errno
+import logging
+import socket
+import threading
+import time
+
+import pytest
+from OpenSSL import SSL
+
+from smartthings_local.errors import SessionClosedError
+from smartthings_local.protocol.dtls_session import DtlsCoapSession
+
+_LOGGER_NAME = "smartthings_local.protocol.dtls_session"
+
+
+class _NullAuth:
+    """Structural AuthenticationProvider — never configured, we skip connect()."""
+
+    def configure_context(self, _context):
+        return None
+
+
+class _FakeConn:
+    """Minimal stand-in for SSL.Connection: each datagram written to the
+    BIO surfaces as one decrypted packet on the next recv(), then
+    WantReadError like a drained DTLS record buffer."""
+
+    def __init__(self):
+        self._decrypted = []
+
+    def bio_write(self, datagram):
+        self._decrypted.append(datagram)
+
+    def recv(self, _n):
+        if self._decrypted:
+            return self._decrypted.pop(0)
+        raise SSL.WantReadError()
+
+    def bio_read(self, _n):
+        return b""
+
+    def send(self, _datagram):
+        return None
+
+    def shutdown(self):
+        return None
+
+
+class _FakeSock:
+    """Scripted UDP socket. Each step is bytes to return, an exception to
+    raise, or a callable to run (then a timeout, so the loop re-checks
+    _stop). An exhausted script blocks like a real recv timeout; once
+    close()d it raises EBADF the way a closed fd does."""
+
+    def __init__(self, steps=()):
+        self._steps = list(steps)
+        self.closed = False
+        self.timeout = None
+
+    def settimeout(self, value):
+        self.timeout = value
+
+    def recv(self, _n):
+        if self.closed:
+            raise OSError(errno.EBADF, "bad file descriptor")
+        if self._steps:
+            step = self._steps.pop(0)
+            if callable(step):
+                step()
+                raise socket.timeout()
+            if isinstance(step, BaseException):
+                raise step
+            return step
+        time.sleep(0.01)
+        raise socket.timeout()
+
+    def send(self, data):
+        return len(data)
+
+    def close(self):
+        self.closed = True
+
+
+def _make_session():
+    sess = DtlsCoapSession("host", 1234, auth=_NullAuth())
+    sess.conn = _FakeConn()
+    return sess
+
+
+def _run_reader(sess, steps, timeout=2.0):
+    sess.sock = _FakeSock(steps)
+    sess.start_reader()
+    sess._reader_thread.join(timeout)
+    assert not sess._reader_thread.is_alive(), "reader thread did not exit"
+
+
+def test_advisory_icmp_error_does_not_kill_reader(caplog):
+    sess = _make_session()
+    dispatched = []
+    sess._dispatch_coap = dispatched.append
+
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        _run_reader(sess, [
+            OSError(errno.ECONNREFUSED, "connection refused"),
+            b"\x60\x00\x00\x00",              # survives, gets dispatched
+            lambda: sess._stop.set(),         # end the loop cleanly
+        ])
+
+    assert dispatched == [b"\x60\x00\x00\x00"]
+    assert not sess._reader_running.is_set()
+    assert any(r.levelno == logging.DEBUG and "advisory" in r.getMessage()
+               for r in caplog.records)
+    # An advisory errno is not a real exit — no WARNING.
+    assert not any(r.levelno >= logging.WARNING for r in caplog.records)
+
+
+def test_fatal_socket_error_exits_with_warning(caplog):
+    sess = _make_session()
+
+    with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
+        _run_reader(sess, [OSError(errno.EBADF, "bad file descriptor")])
+
+    assert not sess._reader_running.is_set()
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1
+    assert "reader exiting" in warnings[0].getMessage()
+
+
+def test_request_fails_fast_after_reader_death():
+    sess = _make_session()
+    _run_reader(sess, [OSError(errno.EBADF, "bad file descriptor")])
+    assert not sess._reader_running.is_set()
+
+    start = time.monotonic()
+    with pytest.raises(SessionClosedError):
+        sess.get(["oic", "d"], timeout=10.0)
+    elapsed = time.monotonic() - start
+    # The whole point: no waiting out the request timeout.
+    assert elapsed < 1.0, f"get() waited {elapsed:.2f}s instead of failing fast"
+
+
+def test_close_does_not_log_warning_on_teardown(caplog):
+    sess = _make_session()
+    sess.sock = _FakeSock()          # empty script: blocks on recv
+    sess.start_reader()
+    time.sleep(0.05)                 # let the reader reach recv
+
+    with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
+        sess.close()
+        sess._reader_thread.join(2.0)
+
+    assert not sess._reader_thread.is_alive()
+    assert not sess._reader_running.is_set()
+    assert not any(r.levelno >= logging.WARNING for r in caplog.records)
+
+
+def test_check_live_without_reader_matches_old_conn_guard():
+    sess = _make_session()          # conn set, reader never started
+    assert sess._reader_thread is None
+    sess._check_live()              # must not raise — config-flow behaviour
+
+    sess.conn = None
+    with pytest.raises(SessionClosedError):
+        sess._check_live()