From eadb2f9580a1e44998bca112312f4fecd86de9ad Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Sat, 29 Aug 2026 10:45:17 -0700 Subject: [PATCH] Fix Observe delivery and two-phase cleanup --- README.md | 15 ++-- smartthings_local/protocol/dtls_session.py | 86 +++++++++++++++++--- tests/test_observe_operations.py | 54 ++++++++++++- tests/test_observe_relations.py | 92 +++++++++++++++++++++- tests/test_session_shutdown.py | 56 ++++++++++++- 5 files changed, 272 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 0eae7d1..28edf6f 100644 --- a/README.md +++ b/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 diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index ade3e0f..fa53924 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/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 diff --git a/tests/test_observe_operations.py b/tests/test_observe_operations.py index f856f9d..bfe230a 100644 --- a/tests/test_observe_operations.py +++ b/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(): diff --git a/tests/test_observe_relations.py b/tests/test_observe_relations.py index 7c4ffa7..561ee94 100644 --- a/tests/test_observe_relations.py +++ b/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,8 +121,12 @@ 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") @@ -131,6 +134,7 @@ def test_plain_initial_response_needs_different_mid_to_confirm_legacy(): 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(): diff --git a/tests/test_session_shutdown.py b/tests/test_session_shutdown.py index 9dc6bc6..2b0d196 100644 --- a/tests/test_session_shutdown.py +++ b/tests/test_session_shutdown.py @@ -9,6 +9,8 @@ from OpenSSL import SSL from smartthings_local.errors import SessionClosedError +from smartthings_local.protocol import dtls_session +from smartthings_local.protocol.coap import OBSERVE, URI_PATH, URI_QUERY, parse_coap from smartthings_local.protocol.dtls_session import DtlsCoapSession @@ -21,6 +23,11 @@ class _CloseNotifyConnection: def __init__(self): self.shutdown_calls = 0 self._outbound = [] + self.sent = [] + + def send(self, datagram): + self.sent.append(datagram) + return len(datagram) def shutdown(self): self.shutdown_calls += 1 @@ -46,6 +53,9 @@ def send(self, datagram): self.sent.append(datagram) return len(datagram) + def settimeout(self, _timeout): + return None + def close(self): self.closed = True @@ -180,18 +190,31 @@ def test_quiesce_wakes_idle_refetch_worker(): session.close() -def test_close_after_quiesce_flushes_close_notify_without_deregistering(): +def test_close_after_quiesce_deregisters_then_flushes_close_notify(): session = _session() connection = session.conn sock = session.sock session._observe_tokens = {b"o": "/mode/vs/0"} - session._send_observe_dereg = lambda *_args: pytest.fail( - "quiesced close sent an application request" - ) + session._observe_queries = {b"o": ("if=oic.if.a",)} + session._pace_orderly_close = lambda: None session.quiesce_for_close() session.close() + assert len(connection.sent) == 1 + _mtype, _code, _mid, token, options, _payload = parse_coap( + connection.sent[0] + ) + assert token == b"o" + assert [value for number, value in options if number == URI_PATH] == [ + b"mode", + b"vs", + b"0", + ] + assert [value for number, value in options if number == URI_QUERY] == [ + b"if=oic.if.a" + ] + assert (OBSERVE, b"\x01") in options assert connection.shutdown_calls == 1 assert sock.sent == [ b"\x15\xfe\xfd\x00\x00" + b"\x00" * 6 @@ -208,6 +231,19 @@ def test_close_after_quiesce_flushes_close_notify_without_deregistering(): assert connection.shutdown_calls == 1 +def test_orderly_close_pacing_waits_after_quiesce(monkeypatch): + session = _session(rate_limit_rps=4) + session._last_send_ts = 10.0 + sleeps = [] + monkeypatch.setattr(dtls_session.time, "monotonic", lambda: 10.1) + monkeypatch.setattr(dtls_session.time, "sleep", sleeps.append) + + session.quiesce_for_close() + session._pace_orderly_close() + + assert sleeps == pytest.approx([0.15]) + + def test_abort_closes_without_close_notify_and_is_idempotent(): session = _session() connection = session.conn @@ -254,3 +290,15 @@ def test_normal_close_still_deregisters_before_quiescing(monkeypatch): session.close() assert calls == [(b"o", ["mode", "vs", "0"], ())] + + +def test_reader_exit_preserves_relations_for_quiesced_close(): + session = _session() + session._observe_tokens = {b"o": "/mode/vs/0"} + session._observe_queries = {b"o": ("if=oic.if.a",)} + + session.quiesce_for_close() + session._reader_loop() + + assert session._observe_tokens == {b"o": "/mode/vs/0"} + assert session._observe_queries == {b"o": ("if=oic.if.a",)}