From 981d0c6e442361f037f1f93555628c16d2f7b6d8 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Sun, 23 Aug 2026 04:15:41 +0000 Subject: [PATCH 1/3] fix(protocol): retransmit and fail fast on the write path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit post() sent its datagram exactly once and then waited on a bare ev.wait(timeout), while get() retransmits every block through _exchange_block. One lost datagram — request or ACK — was therefore an unrecoverable write, while a read absorbed the identical loss silently. That is the report in LocalThings#384: reads keep working, three unrelated resources intermittently do not. Liveness. The bare wait also skipped the slicing _exchange_block uses, so a reader thread dying mid-write burned the caller's whole 8s and reported a device timeout for what was actually a dead session. _wait_for_block is no longer block-specific — it becomes _wait_live, and post() waits through it, so a reader death surfaces as SessionClosedError within one liveness poll. Retransmission, off by default. post() resends the CON up to write_max_attempts times inside the caller's deadline, with §4.2 backoff, pacing every attempt, and the last attempt taking whatever budget is left. The datagram is built once and resent verbatim; reusing the MID is the load-bearing part, because a server implementing §4.5 can then recognise the duplicate and answer from its dedupe cache instead of re-running the write. A caller-side retry cannot offer that — post() mints a fresh MID and token per call, so a retry from above is a genuinely new request the device has no way to dedupe. It defaults to 1 attempt: on that path the wire behaviour is unchanged, one datagram sent in the same order as before, per the ordering caution on #384. Retransmitting into a device already dropping under load turns one lost write into several, and §4.5 dedupe is unverified on RT-OCF, which does not reliably emit RST either. With pacing (#51) landed we can see whether writes are still lost before turning this on, and the flag is then a one-line change. Two details that are not carried over unchanged from the single-send version, both covered by tests: * timeout now bounds the whole call rather than the wait after the send. Attempts share one budget, so it has to be armed before the first pace — and a caller that asked for 8s should not wait 8s plus however long the rate limiter withheld the request. * a retransmission that fails to send is best-effort. A connected UDP socket reports the ICMP error queued by an earlier send on the next one, and the reader already treats those errnos as advisory; failing the exchange there would make retransmitting less robust than leaving it off. Attempt 0 still raises, since it is the caller's only datagram. Rebased onto the shared MID registry: the empty-ACK and RST matching this originally carried is #57's now, and #58 gave the read path the same one-datagram-per-exchange shape, so what remains here is the write attempt loop and the frames that must stop it. Attempt 0 keeps #51's pace-then-check-then-send ordering exactly; the liveness recheck is skipped only once something has answered, because an answer that beat a dying reader is a write the device confirmed and must not be discarded. --- README.md | 25 ++ smartthings_local/protocol/dtls_session.py | 141 +++++++-- tests/test_dtls_session_post_retry.py | 348 +++++++++++++++++++++ 3 files changed, 483 insertions(+), 31 deletions(-) create mode 100644 tests/test_dtls_session_post_retry.py diff --git a/README.md b/README.md index 6527ffb..3d88677 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,31 @@ Setting the signal stops subscribed connection attempts and closes their temporary UDP sockets. It does not alter an already established session or add new session lifecycle methods. Interrupted attempts raise `SessionClosedError`. +Reads retransmit each Block2 request; writes send once. Where a lost write +has been shown to be the cause rather than a device that is simply refusing +load, `write_max_attempts` lets `post()` retransmit inside the caller's own +timeout, backing off per RFC 7252 §4.2 and pacing every retransmit: + +```python +sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth, write_max_attempts=3) +``` + +Each attempt resends the byte-identical datagram, so a server implementing +§4.5 can recognise the duplicate and answer from its dedupe cache instead of +re-running the write. Retrying from the caller cannot do that — a second +`post()` mints a fresh Message ID, which is a new request. It defaults to `1` +(send once) because retransmitting into an appliance that is already dropping +under load turns one lost write into several, and §4.5 dedupe is unverified on +RT-OCF. + +Note that `post()`'s `timeout` bounds the whole call, rate-limit pacing +included, rather than only the wait that follows the send. Every attempt has to +share one budget, and a caller that asked for 8 seconds should not wait 8 +seconds plus however long the limiter withheld the request. At the default 5 +req/s that is at most 200 ms of the budget; at a hand-tuned `rate_limit_rps=1.0` +it is a full second, so a caller pairing a low rate limit with a short timeout +should raise the timeout to match. + If the cert/key are minted at runtime and never written to disk (e.g. inside an HA config flow), create the provider from memory instead: diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index 5c09dc2..5469ced 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -104,6 +104,14 @@ _BLOCK_MAX_ATTEMPTS = 3 _BLOCK_ACK_TIMEOUT = 4.0 +# Base per-attempt wait for a write retransmission, doubled per attempt +# (RFC 7252 §4.2). Retransmission itself is off by default: a device that +# is already dropping under load turns one lost write into several, and +# MID dedupe (§4.5) is unverified on RT-OCF, which does not reliably emit +# RST either. Enable per session via write_max_attempts once pacing has +# been shown insufficient on real hardware (LocalThings#384). +_WRITE_ACK_TIMEOUT = 2.0 + # How often a request wait re-checks that the reader is still alive. Short # enough that a mid-exchange reader death fails fast instead of burning # the whole per-attempt timeout, long enough to stay off the CPU. @@ -250,6 +258,7 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, on_notification=None, mtu=1200, rate_limit_rps: float = _DEFAULT_RATE_LIMIT_RPS, local_port=None, family=socket.AF_UNSPEC, + write_max_attempts: int = 1, auth: AuthenticationProvider | None = None): file_supplied = cert_path is not None or key_path is not None memory_supplied = cert_pem is not None or key_pem is not None @@ -283,6 +292,7 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, self.on_notification = on_notification # fn(href, payload_bytes) self.mtu = mtu self._min_req_interval = 1.0 / rate_limit_rps + self._write_max_attempts = max(1, int(write_max_attempts)) # Optional fixed UDP source port. A client that dies without # close_notify leaves an orphaned DTLS association on the device, # keyed to the old 5-tuple; reconnecting from a fresh ephemeral @@ -1153,7 +1163,17 @@ def _block_num_matches(message, num, szx): 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.""" + (code, payload_bytes). body_cbor must already be encoded. + + timeout bounds the whole call, pacing included, so post() returns + within it rather than within it plus a rate-limit interval. + + Retransmits the CON up to write_max_attempts times within that + deadline, reusing the same Message ID: that is what lets a + server implementing RFC 7252 §4.5 answer a duplicate from its dedupe + cache instead of re-running a non-idempotent write. A caller-side + retry cannot offer that, since it mints a fresh MID and token. + Defaults to one attempt — see _WRITE_ACK_TIMEOUT.""" self._check_live() tok = self._next_tok() opts = [(URI_PATH, s.encode()) for s in path_segs] @@ -1162,38 +1182,97 @@ def post(self, path_segs, body_cbor, timeout=8.0): ev = threading.Event() container = {} mid, exchange = self._register_pending_request(tok, ev, container) + # Built once and resent verbatim: §4.2 defines a retransmission as + # the same message, and a fresh MID would present each retry to the + # appliance as a brand-new write it may run again. + datagram = build_coap(TYPE_CON, METHOD_POST, mid, tok, opts, + body_cbor) + attempts = self._write_max_attempts + # Armed before the first pace, not after the send: every attempt + # shares one budget, and a caller that asked for 8s should not wait + # 8s plus however long the rate limiter withheld the request. + deadline = time.time() + timeout try: - # See get(): a reader can die after the entry check but before - # registration. This post-registration snapshot fails closed. - self.pace() - # The reader can exit between the entry liveness check and the - # registration above. Recheck after registration so its teardown - # cannot miss this waiter. - self._check_live() - datagram = build_coap(TYPE_CON, METHOD_POST, mid, tok, opts, - body_cbor) - self._send_dgram(datagram) - deadline = time.time() + timeout - while True: + for attempt in range(attempts): + self.pace() with self._state_lock: - error = container.get('err') - has_response = 'code' in container - response = ( - (container['code'], container['payload']) - if has_response else None - ) - if error is None and not has_response: - ev.clear() - if error is not None: - raise error - if has_response: - return response - remaining = deadline - time.time() - if remaining <= 0 or not ev.wait(remaining): - raise SessionTimeoutError() - # An empty ACK only stops retransmission. POST does not retry - # today, but keeping the acknowledged exchange pending here is - # the common contract the write retry path will build on. + answered = 'code' in container or 'err' in container + acknowledged = exchange.acknowledged + # Neither can be true before the first send, so attempt 0 is + # byte-for-byte the single send this used to do. On a + # retransmit either one means the datagram arrived: the + # answer landed while we paced, or the device acked it + # separately and owes us only the response. + if not answered: + # The reader can exit between the entry liveness check + # and the registration above, and again during any pace. + # Rechecking here closes that race — but only while + # nothing has answered, because an answer that beat a + # dying reader is a write the device confirmed and must + # not be discarded as a closed session. + self._check_live() + if not acknowledged: + try: + self._send_dgram(datagram) + except EndpointError: + # Attempt 0 is the caller's only datagram, so its + # failure is theirs to see. A retransmit is + # best-effort: a connected UDP socket reports the + # ICMP error queued by an earlier send on the next + # one, and the reader treats those same errnos as + # advisory. Failing the exchange on one would make + # retransmitting less robust than not bothering, + # while the original datagram may still be + # answered inside the budget already running. + if not attempt: + raise + logger.debug( + "POST %s /%s: retransmit %d/%d send failed", + self.host, '/'.join(path_segs), + attempt + 1, attempts, + ) + last = attempt == attempts - 1 + while True: + with self._state_lock: + error = container.get('err') + has_response = 'code' in container + response = ( + (container['code'], container['payload']) + if has_response else None + ) + if error is None and not has_response: + ev.clear() + acknowledged = exchange.acknowledged + if error is not None: + raise error + if has_response: + return response + remaining = deadline - time.time() + if remaining <= 0: + raise SessionTimeoutError() + # An empty ACK stops retransmission (§5.2.2): the device + # took the write and owes only the separate response, so + # spend the rest of the caller's budget waiting for it. + if acknowledged or last: + per_wait = remaining + else: + per_wait = min(_WRITE_ACK_TIMEOUT * (2 ** attempt), + remaining) + if self._wait_live(ev, per_wait): + continue # something moved — re-read the state + if acknowledged or last: + raise SessionTimeoutError() + break # attempt exhausted, retransmit + # Retry only if the next attempt's pace still fits inside the + # caller's deadline: pace() sleeps up to a whole interval, so + # asking merely for "any budget left" returns late. + if deadline - time.time() <= self._min_req_interval: + break + logger.debug( + "POST %s /%s: attempt %d/%d timeout, retransmitting", + self.host, '/'.join(path_segs), attempt + 1, attempts, + ) + raise SessionTimeoutError() finally: self._unregister_pending_request(tok, mid, exchange) diff --git a/tests/test_dtls_session_post_retry.py b/tests/test_dtls_session_post_retry.py new file mode 100644 index 0000000..d46ab1b --- /dev/null +++ b/tests/test_dtls_session_post_retry.py @@ -0,0 +1,348 @@ +"""Write-path retransmission and liveness (LocalThings#384, #396). + +`post()` sent its datagram exactly once and then waited on a bare +`ev.wait(timeout)`, while `get()` retransmitted every block through +`_exchange_block`. One lost datagram was therefore an unrecoverable write +and a silent no-op for a read — three unrelated resources on one AC all +failing with `SessionTimeoutError` is what surfaced it. + +Retransmission ships off by default (`write_max_attempts=1`): a device +already dropping under load turns one lost write into several, and MID +dedupe is unverified on RT-OCF. These tests pin both the default's +unchanged single send and the behaviour the flag buys when it is on. + +The MID registry these lean on is #57's, shared with the read path; #58 +gave `_exchange_block` the same one-datagram-per-exchange shape. What is +specific to writes is the attempt loop: its backoff, its pacing, and the +frames that must stop it. +""" +import threading +import time + +import pytest + +from smartthings_local.errors import ( + EndpointError, + SessionClosedError, + SessionResetError, + SessionTimeoutError, +) +from smartthings_local.protocol import dtls_session as ds +from smartthings_local.protocol.coap import TYPE_ACK, TYPE_RST, parse_coap +from smartthings_local.protocol.dtls_session import DtlsCoapSession + + +class _NullAuth: + """Structural AuthenticationProvider — never configured, we skip connect().""" + + def configure_context(self, _context): + return None + + +def _session(**kwargs): + """Session with the wire stubbed out: every datagram post() hands to + _send_dgram is recorded instead of sent, so a test can decide which + ones the 'device' answers. + + The stub stamps _last_send_ts exactly as the real _send_dgram does -- + without it pace() reads a zero timestamp, decides the interval elapsed + long ago, and never sleeps, which silently voids any test of pacing.""" + sess = DtlsCoapSession("host", 1234, auth=_NullAuth(), **kwargs) + sess.conn = object() # satisfies _check_live's conn guard + sess.sent = [] + + def _record(datagram): + sess.sent.append(datagram) + sess._last_send_ts = time.monotonic() + + sess._send_dgram = _record + return sess + + +def _answer(sess, tok, *, code=0x44, payload=b"", delay=0.0): + """Resolve `tok` the way the reader thread would, optionally late. + + Waits for post() to register the token first — the real reader can + only ever see a token that is already pending.""" + + def _deliver(): + entry = None + give_up = time.monotonic() + 5.0 + while entry is None and time.monotonic() < give_up: + with sess._state_lock: + entry = sess._pending.get(tok) + if entry is None: + time.sleep(0.005) + if entry is None: + return + if delay: + time.sleep(delay) + ev, container = entry + container.update(code=code, payload=payload) + ev.set() + + t = threading.Thread(target=_deliver, daemon=True) + t.start() + return t + + +def _reply_on_second_pace(sess, tok, *, then=None): + """Stub pace() so the *retry's* pace is the window an answer lands in. + + post() paces before its first send too (#51), so a stub that answers on + every call would resolve the exchange before anything reached the wire + and prove nothing about the retransmit.""" + paces = [] + + def _paced(): + paces.append(None) + if len(paces) < 2: + return + with sess._state_lock: + entry = sess._pending.get(tok) + if entry is not None: + ev, container = entry + container.update(code=0x44, payload=b"") + ev.set() + if then is not None: + then() + + sess.pace = _paced + return paces + + +def _control_frame_once_sent(sess, mtype, mid): + """Dispatch a bare ACK/RST for `mid` once the request is on the wire. + + Keyed off `sent` rather than the registry: the exchange is registered + before the send, but a device cannot answer a datagram it has not + received, and firing early would test a case the wire cannot produce.""" + + def _fire(): + give_up = time.monotonic() + 5.0 + while time.monotonic() < give_up: + if sess.sent: + sess._dispatch_coap(ds.build_coap(mtype, 0, mid, b"", [])) + return + time.sleep(0.005) + + threading.Thread(target=_fire, daemon=True).start() + + +def _token_of(sess): + """The token post() will mint next, so a test can answer it.""" + return (sess._tok_counter + 1).to_bytes(4, "big") + + +def _mid_of(sess): + """The MID post() will mint next — what a bare frame is matched on.""" + return (sess._mid + 1) & 0xFFFF + + +def test_default_sends_exactly_once(): + sess = _session() + _answer(sess, _token_of(sess)) + + code, _ = sess.post(["power", "vs", "0"], b"\xa0", timeout=1.0) + + assert code == 0x44 + # The default must stay byte-for-byte the old behaviour: no extra load + # on a device nobody has yet measured as safe to retransmit into. + assert len(sess.sent) == 1 + + +def test_retransmit_recovers_a_dropped_datagram(monkeypatch): + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.1) + sess = _session(write_max_attempts=2, rate_limit_rps=20.0) + # Answers well after the second attempt is on the wire (~0.15s in: a + # 0.1s first attempt plus the 0.05s pace before the retransmit). The + # margin is wide because a loaded CI runner may stall anywhere in there, + # and the assertion is about which attempt gets answered, not when. + _answer(sess, _token_of(sess), delay=1.0) + + code, _ = sess.post(["power", "vs", "0"], b"\xa0", timeout=4.0) + + assert code == 0x44 + assert len(sess.sent) == 2 + + +def test_retransmit_reuses_the_same_message_id_and_token(monkeypatch): + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.1) + sess = _session(write_max_attempts=3) + + with pytest.raises(SessionTimeoutError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=2.0) + + assert len(sess.sent) > 1 + # The whole point of retransmitting rather than re-posting: a server + # implementing RFC 7252 §4.5 can only recognise the duplicate — and skip + # re-running the write — if the frame is the one it already saw. + assert len(set(sess.sent)) == 1, "retransmissions must be byte-identical" + frames = [parse_coap(d) for d in sess.sent] + assert len({f[2] for f in frames}) == 1, "retransmit minted a new MID" + assert len({f[3] for f in frames}) == 1, "retransmit minted a new token" + assert {f[:2] for f in frames} == {(ds.TYPE_CON, ds.METHOD_POST)} + + +def test_attempts_stop_at_the_callers_deadline(monkeypatch): + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.05) + sess = _session(write_max_attempts=10) + + start = time.monotonic() + with pytest.raises(SessionTimeoutError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=0.4) + elapsed = time.monotonic() - start + + # Retransmission lives inside the caller's timeout, never on top of it. + assert elapsed < 1.0, f"post() overran its 0.4s deadline by {elapsed:.2f}s" + + +def test_a_retry_is_skipped_when_its_pace_would_outrun_the_deadline(monkeypatch): + """pace() sleeps up to a whole rate-limit interval, so a retry decided on + "is there any budget left" returns well past the caller's timeout — 1s on + a 0.5s call at 1 rps.""" + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.1) + sess = _session(write_max_attempts=5, rate_limit_rps=1.0) + + start = time.monotonic() + with pytest.raises(SessionTimeoutError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=0.5) + elapsed = time.monotonic() - start + + assert elapsed < 0.5, f"post() overran its 0.5s deadline by {elapsed:.2f}s" + assert len(sess.sent) == 1 + + +def test_a_reply_during_the_pace_window_is_not_resent(monkeypatch): + """The retry's pace is a window the answer can land in. Resending then + puts a second copy of a non-idempotent write on the wire for nothing.""" + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.05) + sess = _session(write_max_attempts=3) + paces = _reply_on_second_pace(sess, _token_of(sess)) + + code, _ = sess.post(["power", "vs", "0"], b"\xa0", timeout=2.0) + + assert code == 0x44 + assert len(paces) == 2, "the reply has to land in the retry's pace window" + assert len(sess.sent) == 1 + + +def test_an_answer_that_beat_a_dying_reader_is_still_returned(monkeypatch): + """That same pace window is one in which both can happen: the response + lands and the reader exits. Checking liveness first throws away a write + the device confirmed and reports the session as closed.""" + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.05) + sess = _session(write_max_attempts=3) + sess._reader_thread = threading.Thread(target=lambda: None) + sess._reader_running.set() + paces = _reply_on_second_pace( + sess, _token_of(sess), then=sess._reader_running.clear + ) + + code, _ = sess.post(["power", "vs", "0"], b"\xa0", timeout=2.0) + + assert code == 0x44 + assert len(paces) == 2 + assert len(sess.sent) == 1 + + +def test_separate_ack_stops_retransmission(monkeypatch): + """An empty ACK means "response coming on its own CON" (RFC 7252 §5.2.2) + and stops the retransmit timer. #57 matches it by MID; what is tested + here is that the write loop then holds instead of resending.""" + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.05) + sess = _session(write_max_attempts=10) + _control_frame_once_sent(sess, TYPE_ACK, _mid_of(sess)) + + with pytest.raises(SessionTimeoutError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=0.5) + + # Without honouring the ACK this retransmits for the whole 0.5s budget. + assert len(sess.sent) == 1 + + +def test_rst_surfaces_as_a_rejection_and_stops_retransmission(monkeypatch): + """RST rejects the request and likewise stops retransmission (§4.2). + Resending through one would push copies of a write the device has + already refused, and then report it as a timeout.""" + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.05) + sess = _session(write_max_attempts=10) + _control_frame_once_sent(sess, TYPE_RST, _mid_of(sess)) + + with pytest.raises(SessionResetError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=1.0) + + assert len(sess.sent) == 1 + + +def test_reader_death_mid_write_fails_fast(): + sess = _session() + sess._reader_thread = threading.Thread(target=lambda: None) + sess._reader_running.set() # alive at entry, so _check_live passes + + def _kill(): + time.sleep(0.05) + sess._reader_running.clear() + + threading.Thread(target=_kill, daemon=True).start() + + start = time.monotonic() + with pytest.raises(SessionClosedError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=10.0) + elapsed = time.monotonic() - start + # Previously this waited out the full timeout on a bare ev.wait() and + # reported it as a device timeout, hiding a dead session behind the + # write's symptom. + assert elapsed < 2.0, f"post() waited {elapsed:.2f}s instead of failing fast" + + +def test_a_failed_retransmit_does_not_abort_the_exchange(monkeypatch): + """A connected UDP socket reports the ICMP error queued by an earlier + send on the *next* one, and the reader treats those errnos as advisory. + If a retransmit's EndpointError killed the exchange, turning + retransmission on would be less robust than leaving it off.""" + monkeypatch.setattr(ds, "_WRITE_ACK_TIMEOUT", 0.1) + sess = _session(write_max_attempts=3, rate_limit_rps=20.0) + tok = _token_of(sess) + attempts = [] + + def _send(datagram): + attempts.append(datagram) + sess._last_send_ts = time.monotonic() + if len(attempts) == 2: + raise EndpointError() # ICMP from attempt 1, surfaced here + + sess._send_dgram = _send + _answer(sess, tok, delay=1.0) # lands well after the failed retry + + assert sess.post(["power", "vs", "0"], b"\xa0", timeout=4.0) == (0x44, b"") + assert len(attempts) >= 2 + + +def test_a_failed_first_send_still_reaches_the_caller(): + """The other side of that: attempt 0 is the caller's only datagram, so + its failure is theirs to see, not something to swallow and time out on.""" + sess = _session(write_max_attempts=3) + + def _send(_datagram): + raise EndpointError() + + sess._send_dgram = _send + + with pytest.raises(EndpointError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=2.0) + + +def test_the_caller_timeout_bounds_the_pace_too(): + """post() paces before its first send (#51). The deadline is armed + before that pace, so a caller that asked for 0.5s gets an answer or an + error inside 0.5s — not 0.5s plus whatever the rate limiter withheld.""" + sess = _session(rate_limit_rps=2.0) # 500ms interval + sess._last_send_ts = time.monotonic() # a send just went out + + start = time.monotonic() + with pytest.raises(SessionTimeoutError): + sess.post(["power", "vs", "0"], b"\xa0", timeout=0.5) + elapsed = time.monotonic() - start + + assert elapsed < 0.9, f"post() took {elapsed:.2f}s for a 0.5s timeout" From b8049a54b981cc52a96f22f6cceb21a450db153b Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Sun, 23 Aug 2026 04:16:16 +0000 Subject: [PATCH 2/3] fix(protocol): keep a response the reader dispatched as it tore down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _close_pending_requests() stamped SessionClosedError onto every pending container with setdefault(), including one that already held the response the reader had just dispatched. Both callers read 'err' before 'code' — post() in its wait loop, _blockwise_get_once on the container _exchange_block hands back — so a request the device had already answered surfaced as a closed session. The window is the reader's own teardown: _reader_loop's finally clears _reader_running and calls _close_pending_requests(), and a response dispatched in the same pass has not necessarily been picked up by its caller yet. Reproduced on the write path 5/5. Only stamp the error on an exchange that has no answer, mirroring the guard the RST branch already applies. Shared by both indices, so the read path gets it too: a final block delivered as the reader exits is an answer, not a closed session. Found reviewing the write-path retransmission that sits under this: its liveness recheck deliberately yields to an answer that beat a dying reader, which the teardown then overwrote anyway. --- smartthings_local/protocol/dtls_session.py | 8 +++- tests/test_dtls_session_post_retry.py | 50 ++++++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index 5469ced..923ba2e 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -611,7 +611,13 @@ def _close_pending_requests(self): ) pending = list(pending_by_id.values()) for _ev, container in pending: - container.setdefault('err', SessionClosedError()) + # Only fail an exchange that has no answer yet. A response + # the reader dispatched before it tore down is a real one — + # the request finished, the session merely died after it — + # and both callers check 'err' before 'code', so stamping + # one here discards a write the device already confirmed. + if 'code' not in container: + container.setdefault('err', SessionClosedError()) self._pending.clear() self._pending_mids.clear() for ev, _container in pending: diff --git a/tests/test_dtls_session_post_retry.py b/tests/test_dtls_session_post_retry.py index d46ab1b..a0f4b7c 100644 --- a/tests/test_dtls_session_post_retry.py +++ b/tests/test_dtls_session_post_retry.py @@ -296,6 +296,56 @@ def _kill(): assert elapsed < 2.0, f"post() waited {elapsed:.2f}s instead of failing fast" +def test_a_response_delivered_as_the_reader_tore_down_is_not_lost(): + """The liveness guard above covers the reader exiting during a pace. The + other half is the reader's own teardown: _reader_loop's finally calls + _close_pending_requests(), which stamped a closed-session error onto + every pending container — including one already holding the response it + had just dispatched. Both callers read 'err' before 'code', so a write + the device confirmed came back as SessionClosedError.""" + sess = _session() + sess._reader_thread = threading.Thread(target=lambda: None) + sess._reader_running.set() + tok = _token_of(sess) + + def _deliver_then_tear_down(_datagram): + sess.sent.append(_datagram) + + def _reader_exit(): + with sess._state_lock: + ev, container = sess._pending[tok] + container.update(code=0x44, payload=b"ok") + ev.set() + # _reader_loop's finally, in order. + sess._reader_running.clear() + sess._close_pending_requests() + + thread = threading.Thread(target=_reader_exit) + thread.start() + thread.join() # teardown wins the race, deterministically + + sess._send_dgram = _deliver_then_tear_down + + assert sess.post(["power", "vs", "0"], b"\xa0", timeout=5.0) == (0x44, b"ok") + + +def test_teardown_does_not_overwrite_an_answered_exchange(): + """The same guard, at its source: shared by post() and _exchange_block, + so the read path gets it too — a final block dispatched as the reader + exits is an answer, not a closed session.""" + sess = _session() + answered_ev, answered = threading.Event(), {"code": 0x45, "payload": b"ok"} + silent_ev, silent = threading.Event(), {} + sess._register_pending_request(b"answered", answered_ev, answered) + sess._register_pending_request(b"silent", silent_ev, silent) + + sess._close_pending_requests() + + assert "err" not in answered, "teardown discarded a dispatched response" + assert isinstance(silent["err"], SessionClosedError) + assert answered_ev.is_set() and silent_ev.is_set() + + def test_a_failed_retransmit_does_not_abort_the_exchange(monkeypatch): """A connected UDP socket reports the ICMP error queued by an earlier send on the *next* one, and the reader treats those errnos as advisory. From 0e5bbdfbd8a69bd24e85afa82d3700d6f25d5f6a Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Sun, 23 Aug 2026 04:16:51 +0000 Subject: [PATCH 3/3] fix(protocol): pace the refresh_observes dereg sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit refresh_observes() dropped every OBSERVE registration in a tight loop. Unlike the teardown dereg in close(), which wants out quickly and leaves a session nobody will use again, this one runs against a session that has to keep working afterwards — and an unpaced OBSERVE burst is what wedges an appliance until something forces a new session (LocalThings#396). The two sleeps that stood in for pacing are gone with it. subscribe() paces its own send since #51, so the 50ms between registrations was always shorter than the wait that followed it — the same redundancy #59 removed from the bridge's registration loop, at the sibling call site. The 100ms between the two sweeps is subsumed the same way, by the pace inside the first subscribe(). Note this does not fix the connect-time OBSERVE burst in #396 on its own: that path is the bridge's registration loop, which #51 already paces. --- smartthings_local/protocol/dtls_session.py | 9 +++++++-- tests/test_dtls_session_post_retry.py | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index 923ba2e..d362100 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -1312,18 +1312,23 @@ def refresh_observes(self, paths): old token gets dropped as 'stale' — acceptable for a 6h-scale safety net.""" self._check_live() + # Paced, unlike the teardown dereg in close(): that one wants out + # quickly and the session is finished either way, while this one + # runs against a session that has to keep working afterwards, and an + # unpaced OBSERVE burst is what wedges an appliance until something + # forces a new session (LocalThings#396). The subscribe sweep below + # needs nothing here — subscribe() paces its own send. for tok, href in list(self._observe_tokens.items()): segs = [s for s in href.split('/') if s] try: + self.pace() self._send_observe_dereg(tok, segs) except Exception as e: logger.warning("refresh dereg %s: %s", href, e) self._observe_tokens.clear() - time.sleep(0.1) for path in paths: try: self.subscribe(list(path)) - time.sleep(0.05) except Exception as e: logger.warning("refresh subscribe %s: %s", path, e) diff --git a/tests/test_dtls_session_post_retry.py b/tests/test_dtls_session_post_retry.py index a0f4b7c..87743ef 100644 --- a/tests/test_dtls_session_post_retry.py +++ b/tests/test_dtls_session_post_retry.py @@ -396,3 +396,20 @@ def test_the_caller_timeout_bounds_the_pace_too(): elapsed = time.monotonic() - start assert elapsed < 0.9, f"post() took {elapsed:.2f}s for a 0.5s timeout" + + +def test_refresh_observes_paces_the_dereg_sweep(): + sess = _session() + sess._observe_tokens = {b"\x41": "/power/vs/0", b"\x42": "/oven/vs/0"} + calls = [] + sess.pace = lambda: calls.append("pace") + sess._send_observe_dereg = lambda *_a: calls.append("dereg") + sess.subscribe = lambda *_a: calls.append("subscribe") + + sess.refresh_observes([("power", "vs", "0"), ("oven", "vs", "0")]) + + # This dereg runs against a session that has to keep working afterwards, + # so the burst is paced (LocalThings#396). The subscribe sweep is not + # paced here on purpose — subscribe() is where that belongs, and #51 put + # it there; the sleep that used to stand in for it is gone with #59's. + assert calls == ["pace", "dereg", "pace", "dereg", "subscribe", "subscribe"]