fix(protocol): retransmit and fail fast on the write path

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:
 

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
@@ -601,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:
@@ -1153,7 +1169,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 +1188,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)
 
@@ -1227,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)