Fix probationary Observe delivery and quiesced cleanup

README.md

@@ -81,9 +81,11 @@ An RFC 7641 relation is confirmed only by a valid Observe response option;
 duplicate and stale 24-bit sequence values are not delivered. Some older
 Samsung firmware omits that option. For those devices, a plain initial `2.05`
 is probationary until a later packet arrives on the same token with a different
-Message ID. Optional `on_observe_pending`, `on_legacy_notification`, and
-`on_observe_error` constructor callbacks let consumers keep that compatibility
-path distinct from confirmed RFC notifications and ordinary polling.
+Message ID. Its complete representation is still delivered through
+`on_notification`; a blockwise representation is re-read before delivery.
+Optional `on_observe_pending`, `on_legacy_notification`, and `on_observe_error`
+constructor callbacks let consumers keep that compatibility path distinct from
+confirmed RFC notifications and ordinary polling.
 
 Periodic renewal can target only the relations that need it; unrelated
 observations remain active. Existing query variants are preserved unless the
@@ -143,9 +145,10 @@ Interrupted attempts raise `SessionClosedError`.
 Hosts that stop network work before their blocking executor drains can use the
 session's two-phase shutdown. `quiesce_for_close()` is terminal: it interrupts
 an in-progress handshake, wakes pending requests and notification refetches,
-and rejects new work while retaining an established DTLS socket. A subsequent
-`close()` flushes the authenticated close-notify record before closing that
-socket:
+and rejects new work while retaining an established DTLS socket and active
+Observe relation metadata. A subsequent `close()` paces explicit Observe
+deregistrations, flushes the authenticated close-notify record, and then closes
+that socket:
 
 ```python
 sess.quiesce_for_close()  # safe from the host's early shutdown phase

smartthings_local/protocol/dtls_session.py

@@ -469,6 +469,11 @@ def __init__(self, host, port, cert_path=None, key_path=None, *,
         self._lifecycle_lock = threading.Lock()
 
         self._send_lock = threading.Lock()
+        # Only the thread performing orderly close may send an Observe
+        # deregistration after terminal quiescence. A thread-local exception
+        # keeps concurrent application senders blocked without changing the
+        # existing private send-hook signatures used by test/session adapters.
+        self._orderly_close_send_thread_id = None
         # Guards the MID/token counters and pending-request registries.
         # The refetch worker makes the session its own second concurrent
         # get() caller, so two threads can mint tokens at once; without
@@ -533,6 +538,18 @@ def pace(self) -> None:
         if remaining > 0:
             self._stop.wait(remaining)
 
+    def _pace_orderly_close(self) -> None:
+        """Honor request spacing after terminal quiescence.
+
+        ``quiesce_for_close()`` sets ``_stop`` so ordinary paced work wakes
+        immediately.  A later orderly close still has to space its explicit
+        Observe deregistrations, so this teardown-only path cannot wait on the
+        already-set event.
+        """
+        remaining = self._min_req_interval - (time.monotonic() - self._last_send_ts)
+        if remaining > 0:
+            time.sleep(remaining)
+
     # ---- lifecycle ---------------------------------------------------
 
     def connect(
@@ -712,8 +729,16 @@ def _send_observe_dereg(self, tok, path_segs, query=()):
             opts.append((URI_QUERY, value.encode()))
         opts.append((OBSERVE, OBSERVE_DEREGISTER))
         opts.append((ACCEPT, CF_CBOR))
-        self._send_dgram(
-            build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
+        self._send_dgram(build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
+
+    def _send_observe_dereg_after_quiesce(self, tok, path_segs, query=()):
+        """Permit one orderly-close deregistration on the closing thread."""
+        previous = self._orderly_close_send_thread_id
+        self._orderly_close_send_thread_id = threading.get_ident()
+        try:
+            self._send_observe_dereg(tok, path_segs, query)
+        finally:
+            self._orderly_close_send_thread_id = previous
 
     @staticmethod
     def _send_close_notify(connection, sock):
@@ -764,17 +789,25 @@ def _close_orderly(self):
         # Send dereg for every active observation while the conn is
         # still healthy. Tiny sleep lets the records reach the wire
         # before we shut DTLS down.
-        if (not self._lifecycle_cancel.is_set() and self.conn is not None
-                and self._observe_tokens):
+        if self.conn is not None and self._observe_tokens:
+            quiesced = self._lifecycle_cancel.is_set()
             with self._state_lock:
                 observations = tuple(self._observe_tokens.items())
                 observe_queries = dict(self._observe_queries)
             for tok, href in observations:
                 segs = [s for s in href.split('/') if s]
                 try:
-                    self.pace()
-                    self._send_observe_dereg(
-                        tok, segs, observe_queries.get(tok, ()))
+                    if quiesced:
+                        self._pace_orderly_close()
+                        self._send_observe_dereg_after_quiesce(
+                            tok,
+                            segs,
+                            observe_queries.get(tok, ()),
+                        )
+                    else:
+                        self.pace()
+                        self._send_observe_dereg(
+                            tok, segs, observe_queries.get(tok, ()))
                 except Exception as e:
                     logger.warning("dereg %s: %s", href, e)
             time.sleep(0.1)
@@ -918,7 +951,7 @@ def _clear_observe_relations(self):
             self._observe_sequences.clear()
 
     def _observe_relation_active(self, href, query, legacy):
-        """Return whether one confirmed relation still owns this identity."""
+        """Return whether one relation still owns this callback identity."""
         with self._state_lock:
             for tok, observed_href in self._observe_tokens.items():
                 if observed_href != href or \
@@ -927,7 +960,14 @@ def _observe_relation_active(self, href, query, legacy):
                 if legacy:
                     if tok in self._legacy_observe_tokens:
                         return True
-                elif tok in self._observe_sequences:
+                elif tok in self._observe_sequences or (
+                        tok in self._observe_plain_response_mids
+                        and tok not in self._legacy_observe_tokens):
+                    # A probationary optionless response is not proof of an
+                    # Observe relation, but its complete representation still
+                    # belongs on the ordinary notification callback. Keep a
+                    # Block2 refetch alive until the token is retired or later
+                    # proves the legacy relation.
                     return True
         return False
 
@@ -955,9 +995,17 @@ def _observe_sequence_is_fresh(previous, current, received_at):
 
     def _send_dgram(self, datagram):
         """Send a CoAP datagram. Holds the send lock for the
-        BIO-drain so two writers can't interleave records."""
+        BIO-drain so two writers can't interleave records.
+
+        The orderly-close deregistration helper grants only its calling thread
+        a teardown send after application workers have joined. All ordinary
+        request paths remain blocked once terminal quiescence begins.
+        """
         with self._send_lock:
-            if self._lifecycle_cancel.is_set() or self.conn is None:
+            orderly_close_send = (
+                self._orderly_close_send_thread_id == threading.get_ident())
+            if (self._lifecycle_cancel.is_set() and not orderly_close_send) or \
+                    self.conn is None:
                 raise SessionClosedError()
             send_failed = False
             try:
@@ -1058,7 +1106,12 @@ def _reader_loop(self):
             with self._refetch_cond:
                 self._refetch_pending.clear()
                 self._refetch_cond.notify_all()
-            self._clear_observe_relations()
+            # Two-phase shutdown retains relation metadata for close(), which
+            # runs after application workers have joined and sends the paced
+            # deregistration sweep. Unexpected reader death has no later
+            # orderly phase and must still retire everything immediately.
+            if not self._lifecycle_cancel.is_set():
+                self._clear_observe_relations()
 
     def _dispatch_coap(self, datagram):
         try:
@@ -1221,7 +1274,14 @@ def _dispatch_coap(self, datagram):
                         except Exception as e:
                             logger.debug(
                                 "observe pending callback %s: %s", href, e)
-                    return
+                    with self._state_lock:
+                        if self._observe_tokens.get(tok) != href or \
+                                self._observe_queries.get(tok, ()) != \
+                                observe_query or \
+                                self._observe_plain_response_mids.get(tok) != \
+                                mid or tok in self._legacy_observe_tokens or \
+                                tok in self._observe_sequences:
+                            return
             # RFC 7959 §2.6: a notification carries only the first block
             # of the representation. Handing the callback a partial CBOR
             # buffer is what #39 was about, so anything with M=1 (or a

tests/test_observe_operations.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from smartthings_local.errors import EndpointError
+from smartthings_local.errors import EndpointError, SessionClosedError
 from smartthings_local.protocol.coap import (
     OBSERVE,
     URI_PATH,
@@ -345,18 +345,64 @@ def test_normal_close_paces_every_exact_deregister_and_clears_state(
     assert session._observe_sequences == {}
 
 
-def test_quiesced_close_skips_deregister_pacing():
+def test_quiesced_close_paces_every_deregister_for_orderly_teardown():
     session = _session()
     session.sock = _Socket()
-    _add_relation(session, ["mode", "vs", "0"])
+    token = _add_relation(session, ["mode", "vs", "0"])
     session.pace = Mock()
+    session._pace_orderly_close = Mock()
     session._send_observe_dereg = Mock()
 
     session.quiesce_for_close()
     session.close()
 
     session.pace.assert_not_called()
-    session._send_observe_dereg.assert_not_called()
+    session._pace_orderly_close.assert_called_once_with()
+    session._send_observe_dereg.assert_called_once_with(
+        token,
+        ["mode", "vs", "0"],
+        (),
+    )
+
+
+def test_quiesced_close_preserves_existing_send_override_signature():
+    session = _session()
+    session.sock = _Socket()
+    token = _add_relation(session, ["mode", "vs", "0"])
+    sent = []
+    session._send_dgram = sent.append
+    session._pace_orderly_close = Mock()
+
+    session.quiesce_for_close()
+    session.close()
+
+    assert len(sent) == 1
+    assert parse_coap(sent[0])[3] == token
+
+
+def test_orderly_close_send_permission_is_thread_local():
+    session = _session()
+    session.quiesce_for_close()
+    outcomes = []
+
+    def try_application_send():
+        try:
+            session._send_dgram(b"application request")
+        except Exception as error:  # noqa: BLE001 - captured for assertion
+            outcomes.append(error)
+
+    def during_teardown_send(_token, _path, _query):
+        thread = threading.Thread(target=try_application_send)
+        thread.start()
+        thread.join()
+
+    session._send_observe_dereg = during_teardown_send
+    session._send_observe_dereg_after_quiesce(
+        b"o", ["mode", "vs", "0"], ()
+    )
+
+    assert len(outcomes) == 1
+    assert isinstance(outcomes[0], SessionClosedError)
 
 
 def test_reader_exit_clears_all_relation_state():

tests/test_observe_relations.py

@@ -12,7 +12,6 @@
     BLOCK2,
     OBSERVE,
     TYPE_ACK,
-    TYPE_CON,
     TYPE_NON,
     URI_QUERY,
     block_value,
@@ -71,7 +70,7 @@ def test_subscribe_registers_path_and_query_before_immediate_response():
     def send(datagram):
         request = parse_coap(datagram)
         requests.append(request)
-        _mtype, _code, mid, token, options, _payload = request
+        _mtype, _code, mid, token, _options, _payload = request
         assert session._observe_tokens[token] == "/mode/vs/0"
         assert session._observe_queries[token] == (
             "if=oic.if.a",
@@ -122,15 +121,20 @@ def test_plain_initial_response_needs_different_mid_to_confirm_legacy():
     _notify(session, token, 10, b"initial-retransmit")
 
     assert pending == ["/doors/vs/0"]
-    assert delivered == []
+    assert delivered == [
+        ("standard", "/doors/vs/0", b"initial"),
+    ]
+    assert session._observe_plain_response_mids[token] == 10
     assert token not in session._legacy_observe_tokens
+    assert token not in session._observe_sequences
 
     _notify(session, token, 11, b"changed")
     _notify(session, token, 11, b"changed-retransmit")
     _notify(session, token, 12, b"changed-again")
 
     assert token in session._legacy_observe_tokens
     assert delivered == [
+        ("standard", "/doors/vs/0", b"initial"),
         ("legacy", "/doors/vs/0", b"changed"),
         ("legacy", "/doors/vs/0", b"changed-again"),
     ]
@@ -147,7 +151,87 @@ def test_legacy_notification_falls_back_to_main_callback():
     _notify(session, token, 20, b"initial")
     _notify(session, token, 21, b"changed")
 
-    assert delivered == [("/doors/vs/0", b"changed")]
+    assert delivered == [
+        ("/doors/vs/0", b"initial"),
+        ("/doors/vs/0", b"changed"),
+    ]
+
+
+def test_probationary_blockwise_initial_queues_refetch_without_partial_delivery():
+    delivered = []
+    pending = []
+    session = _session(
+        on_notification=lambda href, payload: delivered.append((href, payload)),
+        on_observe_pending=pending.append,
+    )
+    session._send_dgram = Mock()
+    session._queue_refetch = Mock()
+    token = session.subscribe(
+        ["mode", "vs", "0"], query=("if=oic.if.a",)
+    )
+
+    _notify(
+        session,
+        token,
+        10,
+        b"partial",
+        options=((BLOCK2, block_value(0, 1, 6)),),
+    )
+
+    assert pending == ["/mode/vs/0"]
+    assert delivered == []
+    session._queue_refetch.assert_called_once_with(
+        "/mode/vs/0",
+        ("if=oic.if.a",),
+        legacy=False,
+    )
+
+
+def test_pending_callback_can_retire_relation_before_initial_delivery():
+    delivered = []
+    holder = {}
+
+    def retire(_href):
+        with holder["session"]._state_lock:
+            holder["session"]._retire_observe_token_locked(holder["token"])
+
+    session = _session(
+        on_notification=lambda href, payload: delivered.append((href, payload)),
+        on_observe_pending=retire,
+    )
+    holder["session"] = session
+    session._send_dgram = Mock()
+    holder["token"] = session.subscribe(["mode", "vs", "0"])
+
+    _notify(session, holder["token"], 10, b"initial")
+
+    assert delivered == []
+    assert holder["token"] not in session._observe_tokens
+
+
+def test_probationary_blockwise_representation_is_refetched_before_delivery():
+    delivered = []
+    session = _session(
+        on_notification=lambda href, payload: delivered.append((href, payload))
+    )
+    session._blockwise_get = Mock(
+        return_value=(0x45, b"complete", 2, b"fresh")
+    )
+    token = b"\x41"
+    href = "/mode/vs/0"
+    query = ("if=oic.if.a",)
+    session._observe_tokens[token] = href
+    session._observe_queries[token] = query
+    session._observe_plain_response_mids[token] = 10
+
+    session._refetch_one((href, query, False), 1)
+
+    session._blockwise_get.assert_called_once_with(
+        ["mode", "vs", "0"],
+        query,
+        dtls_session._REFETCH_TIMEOUT_S,
+    )
+    assert delivered == [(href, b"complete")]
 
 
 def test_rejected_observe_retires_every_relation_index():