Merge pull request #40 from QuiteYellow/fix/observe-block2-truncation

fix(protocol): reassemble blockwise OBSERVE notifications

smartthings_local/protocol/coap.py

@@ -13,6 +13,7 @@ from ..errors import MalformedMessageError
 URI_PATH       = 11
 URI_QUERY      = 15
 OBSERVE        =  6
+ETAG           =  4
 CONTENT_FORMAT = 12
 ACCEPT         = 17
 BLOCK2         = 23
@@ -121,6 +122,14 @@ def block_value(num, more, szx):
     return struct.pack('>I', v)[1:]
 
 
+def block_fields(value):
+    """Decode a CoAP Block-N option value. Inverse of block_value().
+    Returns (num, more, szx). An empty value means block 0, no more,
+    SZX=0 — RFC 7959 §2.2 allows a zero-length option to elide it."""
+    v = int.from_bytes(value, 'big')
+    return v >> 4, (v >> 3) & 1, v & 0x07
+
+
 def fmt_code(c):
     """0x45 → '2.05', 0x84 → '4.04'. Used in log lines."""
     return f"{c >> 5}.{c & 0x1F:02d}"

smartthings_local/protocol/dtls_session.py

@@ -12,11 +12,18 @@ Wire-level details that matter (from local-tools/oven-findings.md §17):
     or interleaved one-shot / OBSERVE traffic mis-attributes.
   * Multi-block GET requires the SAME CoAP token across every block
     of the response ("token-stable Block2"). Fresh-token-per-block
-    is silently dropped by the server.
+    is silently dropped by the server, and a transfer that opens at
+    NUM>0 under a token the server has not seen gets no reply at all.
 
 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.
+
+A notification carries only the first block of a large representation
+(RFC 7959 §2.6). Because a continuation cannot borrow the observation's
+token (§3.4) and this server will not continue a transfer it did not
+start, such a notification is withheld and the resource is re-read from
+block 0 on a fresh one-shot token by a worker thread. See #39.
 """
 import errno
 import math
@@ -35,11 +42,12 @@ from ..errors import (
     SessionTimeoutError,
 )
 from .coap import (
-    URI_PATH, URI_QUERY, OBSERVE, CONTENT_FORMAT, ACCEPT, BLOCK2, SIZE2,
+    URI_PATH, URI_QUERY, OBSERVE, ETAG, CONTENT_FORMAT, ACCEPT, BLOCK2, SIZE2,
     TYPE_CON, TYPE_NON, TYPE_ACK, TYPE_RST,
     METHOD_GET, METHOD_POST, CF_CBOR,
     OBSERVE_REGISTER, OBSERVE_DEREGISTER, BLOCK_SZX,
-    encode_options, parse_coap, build_coap, block_value, fmt_code,
+    encode_options, parse_coap, build_coap, block_value, block_fields,
+    fmt_code,
     split_dtls as _split_dtls,
 )
 from .auth import (
@@ -72,6 +80,11 @@ DEBUG_BRIDGE = os.environ.get('DEBUG_BRIDGE') == '1'
 _BLOCK_MAX_ATTEMPTS = 3
 _BLOCK_ACK_TIMEOUT  = 4.0
 
+# How often a block wait re-checks that the reader is still alive. Short
+# enough that a mid-transfer reader death fails fast instead of burning
+# the whole per-block timeout, long enough to stay off the CPU.
+_BLOCK_LIVENESS_POLL_S = 0.25
+
 # Inter-request pacing: minimum seconds between CoAP CON sends on one session.
 # Samsung's RT-OCF stacks drop requests when hit faster than their firmware
 # ceiling (dryer ~14 req/s, oven ~8 req/s, dishwasher unknown). 5 req/s
@@ -79,6 +92,22 @@ _BLOCK_ACK_TIMEOUT  = 4.0
 # once the ceiling is measured empirically.
 _DEFAULT_RATE_LIMIT_RPS = 5.0
 
+# Maximum hrefs held for OBSERVE refetch at once. A notification storm
+# on more resources than this is already past what the 5/s ceiling can
+# drain, so the excess is dropped rather than queued indefinitely.
+_MAX_PENDING_REFETCH = 16
+
+# Timeout for one notification refetch. Generous relative to a poll:
+# the resource is known large (that is why it blocked) and the worker
+# is serialized, so a slow one delays only later refetches.
+_REFETCH_TIMEOUT_S = 15.0
+
+
+class _EtagChanged(Exception):
+    """Internal: the server's ETag changed partway through a Block2
+    transfer, so the blocks in hand are from two different versions."""
+
+
 # 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
@@ -241,6 +270,11 @@ class DtlsCoapSession:
         self.endpoint = None
 
         self._send_lock = threading.Lock()
+        # Guards the MID/token counters and _pending. The refetch worker
+        # makes the session its own second concurrent get() caller, so
+        # two threads can mint tokens at once; without this they can
+        # collide and one transfer silently absorbs the other's blocks.
+        self._state_lock = threading.Lock()
         # Randomize MID and token counter starting points so reconnects
         # don't reuse identifiers from previous sessions — Samsung's
         # RT-OCF appears to remember observer state across DTLS
@@ -257,6 +291,14 @@ class DtlsCoapSession:
         # token (bytes) → href (str)
         self._observe_tokens = {}
 
+        # OBSERVE refetch queue: href → sequence number of the newest
+        # notification that asked for it. Drained by a worker thread
+        # because _dispatch_coap cannot block (see _queue_refetch).
+        self._refetch_cond = threading.Condition()
+        self._refetch_pending = {}
+        self._refetch_seq = 0
+        self._refetch_thread = None
+
         self._stop = threading.Event()
         self._reader_thread = None
         # Set while the reader owns the socket. Cleared when it exits for
@@ -407,6 +449,8 @@ class DtlsCoapSession:
         """Block until the reader thread exits (i.e. socket dies)."""
         if self._reader_thread is not None:
             self._reader_thread.join()
+        if self._refetch_thread is not None:
+            self._refetch_thread.join()
 
     def _send_observe_dereg(self, tok, path_segs):
         """Send a single OBSERVE deregister GET (Observe option = 1)
@@ -438,6 +482,11 @@ class DtlsCoapSession:
             time.sleep(0.1)
 
         self._stop.set()
+        # Wake the refetch worker so it sees _stop instead of sitting on
+        # its condition for up to a second after the socket is gone.
+        with self._refetch_cond:
+            self._refetch_pending.clear()
+            self._refetch_cond.notify_all()
         if self.conn is not None:
             try:
                 self.conn.shutdown()
@@ -448,10 +497,12 @@ class DtlsCoapSession:
                 self.sock.close()
             except Exception:
                 pass
-        for tok, (ev, container) in list(self._pending.items()):
+        with self._state_lock:
+            pending = list(self._pending.items())
+            self._pending.clear()
+        for tok, (ev, container) in pending:
             container.setdefault('err', SessionClosedError())
             ev.set()
-        self._pending.clear()
         self._observe_tokens.clear()
         self.sock = None
         self.conn = None
@@ -461,27 +512,30 @@ class DtlsCoapSession:
     # ---- send / receive plumbing -------------------------------------
 
     def _next_mid(self):
-        self._mid = (self._mid + 1) & 0xFFFF
-        return self._mid
+        with self._state_lock:
+            self._mid = (self._mid + 1) & 0xFFFF
+            return self._mid
 
     def _next_tok(self):
-        self._tok_counter = (self._tok_counter + 1) & 0xFFFFFFFF
-        # 4-byte tokens — fits within tkl=8 cap with headroom and
-        # avoids collisions across long-running OBSERVE subscriptions.
-        return self._tok_counter.to_bytes(4, 'big')
+        with self._state_lock:
+            self._tok_counter = (self._tok_counter + 1) & 0xFFFFFFFF
+            # 4-byte tokens — fits within tkl=8 cap with headroom and
+            # avoids collisions across long-running OBSERVE subscriptions.
+            return self._tok_counter.to_bytes(4, 'big')
 
     def _next_observe_tok(self):
-        # Single-byte tokens for OBSERVE registrations. Samsung
-        # RT-OCF accepts these but silently drops TKL=4 OBSERVE
-        # registrations. Counter is randomly seeded per session so
-        # reconnects don't collide with stale observer state Samsung
-        # may still be holding from the previous run.
-        self._observe_tok_counter = (self._observe_tok_counter + 1) & 0xFF
-        # Avoid 0x00 — some CoAP stacks treat an all-zero token as
-        # equivalent to "no token" / empty (TKL=0).
-        if self._observe_tok_counter == 0:
-            self._observe_tok_counter = 1
-        return bytes([self._observe_tok_counter])
+        with self._state_lock:
+            # Single-byte tokens for OBSERVE registrations. Samsung
+            # RT-OCF accepts these but silently drops TKL=4 OBSERVE
+            # registrations. Counter is randomly seeded per session so
+            # reconnects don't collide with stale observer state Samsung
+            # may still be holding from the previous run.
+            self._observe_tok_counter = (self._observe_tok_counter + 1) & 0xFF
+            # Avoid 0x00 — some CoAP stacks treat an all-zero token as
+            # equivalent to "no token" / empty (TKL=0).
+            if self._observe_tok_counter == 0:
+                self._observe_tok_counter = 1
+            return bytes([self._observe_tok_counter])
 
     def _send_dgram(self, datagram):
         """Send a CoAP datagram. Holds the send lock for the
@@ -583,9 +637,15 @@ class DtlsCoapSession:
             # 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()):
+            with self._state_lock:
+                pending = list(self._pending.items())
+            for tok, (ev, container) in pending:
                 container.setdefault('err', SessionClosedError())
                 ev.set()
+            # Nothing will answer a refetch now either.
+            with self._refetch_cond:
+                self._refetch_pending.clear()
+                self._refetch_cond.notify_all()
 
     def _dispatch_coap(self, datagram):
         try:
@@ -615,7 +675,8 @@ class DtlsCoapSession:
             return
 
         # Pending one-shot? Resolve and return.
-        rec = self._pending.get(tok)
+        with self._state_lock:
+            rec = self._pending.get(tok)
         if rec is not None:
             ev, container = rec
             container['code']    = code
@@ -633,6 +694,16 @@ class DtlsCoapSession:
                 logger.warning("observe %s: non-2.05 %s",
                                href, fmt_code(code))
                 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
+            # block past the first) goes to the refetch worker instead.
+            b2 = [v for n, v in ropts if n == BLOCK2]
+            if b2:
+                num, more, _ = block_fields(b2[0])
+                if more or num:
+                    self._queue_refetch(href)
+                    return
             cb = self.on_notification
             if cb is not None:
                 try:
@@ -644,6 +715,110 @@ class DtlsCoapSession:
 
         # Stale token (post-reconnect or unknown) — drop quietly.
 
+    # ---- OBSERVE refetch ---------------------------------------------
+
+    @staticmethod
+    def _log_refetch(msg, *args):
+        """Refetch outcomes are debug-level in normal operation, which is
+        below the bridge's INFO default, so a healthy session stays quiet.
+        DEBUG_BRIDGE=1 promotes them to INFO for hardware validation:
+        that shows which token the re-read used and whether it completed,
+        without also turning on every per-block retransmit line."""
+        (logger.info if DEBUG_BRIDGE else logger.debug)(msg, *args)
+
+    def _queue_refetch(self, href):
+        """Queue a blockwise notification for re-reading.
+
+        Called from the reader thread, so it must not block: _dispatch_coap
+        runs there and _blockwise_get waits on an Event only that same
+        thread can set, which would deadlock the session outright. Latest
+        wins per href — a burst of notifications on one resource collapses
+        into a single re-read of its final state."""
+        with self._refetch_cond:
+            if (href not in self._refetch_pending
+                    and len(self._refetch_pending) >= _MAX_PENDING_REFETCH):
+                self._log_refetch(
+                    "refetch %s dropped: queue full (%d pending)",
+                    href, len(self._refetch_pending))
+                return
+            self._refetch_seq += 1
+            self._refetch_pending[href] = self._refetch_seq
+            self._refetch_cond.notify()
+        self._start_refetch_worker()
+
+    def _start_refetch_worker(self):
+        """Start the refetch worker on first use. Sessions that never see
+        a blockwise notification never grow the thread."""
+        if self._refetch_thread is not None or self._stop.is_set():
+            return
+        with self._state_lock:
+            if self._refetch_thread is not None or self._stop.is_set():
+                return
+            self._refetch_thread = threading.Thread(
+                target=self._refetch_loop, daemon=True,
+                name=f'stl-refetch-{self.host}')
+            self._refetch_thread.start()
+
+    def _refetch_loop(self):
+        """Re-read blockwise-notified resources, one at a time.
+
+        Serialized on purpose. Each re-read is a multi-block transfer and
+        _blockwise_get paces between blocks, so running one at a time is
+        what keeps a notification storm under the firmware's request
+        ceiling."""
+        while self._refetch_alive():
+            with self._refetch_cond:
+                while not self._refetch_pending and self._refetch_alive():
+                    self._refetch_cond.wait(1.0)
+                if not self._refetch_pending:
+                    return
+                href, seq = next(iter(self._refetch_pending.items()))
+                del self._refetch_pending[href]
+            self._refetch_one(href, seq)
+
+    def _refetch_alive(self):
+        """False once the session is closing or the reader has died. A
+        refetch needs the reader to resolve its token, so outliving it
+        would leave join() waiting on a thread with nothing to do."""
+        if self._stop.is_set():
+            return False
+        return self._reader_thread is None or self._reader_running.is_set()
+
+    def _refetch_one(self, href, seq):
+        """Re-read one href from block 0 and deliver it if it is still
+        the freshest thing we know about that resource."""
+        self.pace()
+        segs = [s for s in href.split('/') if s]
+        try:
+            code, payload, blocks, tok = self._blockwise_get(
+                segs, (), _REFETCH_TIMEOUT_S)
+        except Exception as e:
+            # Device silent, session gone, ETag never settled, block cap
+            # hit. Whatever the reason, dropping the notification is the
+            # contract: the poll tiers still carry freshness, and handing
+            # over the first block is the bug this replaced.
+            self._log_refetch("refetch %s failed: %s", href, e)
+            return
+        if code != 0x45:
+            self._log_refetch("refetch %s returned %s", href, fmt_code(code))
+            return
+        with self._refetch_cond:
+            # A newer notification landed while we were reading. That one
+            # has its own refetch queued, so this result is already stale.
+            if self._refetch_pending.get(href, 0) > seq:
+                self._log_refetch(
+                    "refetch %s tok=%s blocks=%d bytes=%d superseded",
+                    href, tok.hex(), blocks, len(payload))
+                return
+        self._log_refetch("refetch %s tok=%s blocks=%d bytes=%d ok",
+                          href, tok.hex(), blocks, len(payload))
+        cb = self.on_notification
+        if cb is not None:
+            try:
+                cb(href, payload)
+            except Exception as e:
+                logger.warning("notification callback %s: %s", href, e)
+
     # ---- request primitives ------------------------------------------
 
     def get(self, path_segs, query=(), timeout=10.0):
@@ -654,76 +829,183 @@ class DtlsCoapSession:
         token, and dropping a fresh token on block 1+ silently drops
         the request."""
         self._check_live()
+        code, blob, _blocks, _tok = self._blockwise_get(
+            path_segs, query, timeout)
+        return code, blob
+
+    def _blockwise_get(self, path_segs, query=(), timeout=10.0):
+        """Shared token-stable Block2 reassembly (RFC 7959 §2.4).
+
+        Returns (code, payload, block_count, token). The last two are
+        diagnostics for the refetch log; get() drops them.
+
+        Mints one fresh 4-byte token and holds it across every block of
+        the transfer. Also the notification-refetch primitive: RFC 7959
+        §3.4 forbids continuing a blockwise notification on the
+        observation's token, and Samsung's RT-OCF drops a transfer that
+        opens at NUM>0 under a token it has not seen, so a truncated
+        notification is recovered by re-reading from block 0 through
+        this same path rather than by a §2.6 continuation.
+
+        Restarts once if the server's ETag changes mid-transfer, then
+        gives up: RFC 7959 §2.4 requires the client to compare ETags
+        when the server supplies them. None of the tested appliances
+        emit option 4, so on those this is inert."""
+        try:
+            return self._blockwise_get_once(path_segs, query, timeout)
+        except _EtagChanged:
+            logger.debug("GET %s /%s: ETag changed mid-transfer, restarting",
+                         self.host, '/'.join(path_segs))
+        try:
+            return self._blockwise_get_once(path_segs, query, timeout)
+        except _EtagChanged:
+            logger.debug(
+                "GET %s /%s: representation kept changing mid-transfer",
+                self.host, '/'.join(path_segs))
+            raise BlockwiseError() from None
+
+    def _blockwise_get_once(self, path_segs, query, timeout):
+        """One attempt at a full Block2 transfer. Raises _EtagChanged if
+        the server's representation changed while we were reassembling."""
         tok = self._next_tok()
         blob = b''
         num = 0
+        blocks = 0
         last_code = None
-        last_opts = []
+        etag = None
         deadline = time.time() + timeout
         szx = BLOCK_SZX   # server may negotiate down; track per-transfer
         while True:
             if num > 0:
                 self.pace()
-            container = {}
-            for attempt in range(_BLOCK_MAX_ATTEMPTS):
-                ev = threading.Event()
-                container = {}
-                self._pending[tok] = (ev, container)
-                try:
-                    mid = self._next_mid()
-                    opts = [(URI_PATH, s.encode()) for s in path_segs]
-                    for q in query:
-                        opts.append((URI_QUERY, q.encode()))
-                    opts.append((ACCEPT, CF_CBOR))
-                    if num > 0:
-                        opts.append((BLOCK2, block_value(num, 0, szx)))
-                    self._send_dgram(
-                        build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
-                    per_wait = min(_BLOCK_ACK_TIMEOUT,
-                                   max(0.1, deadline - time.time()))
-                    if ev.wait(per_wait):
-                        break  # got a response
-                    remaining = deadline - time.time()
-                    if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1:
-                        logger.debug(
-                            "GET %s /%s block %d: timed out after %d attempt(s)",
-                            self.host, '/'.join(path_segs), num, attempt + 1,
-                        )
-                        raise SessionTimeoutError()
-                    logger.debug(
-                        "GET %s /%s block %d: attempt %d/%d timeout, retrying",
-                        self.host, '/'.join(path_segs), num,
-                        attempt + 1, _BLOCK_MAX_ATTEMPTS,
-                    )
-                finally:
-                    self._pending.pop(tok, None)
+            container = self._exchange_block(
+                tok, path_segs, query, num, szx, deadline)
             if 'err' in container:
                 raise container['err']
+            blocks += 1
 
             code = container['code']
             payload = container['payload']
             ropts = container['options']
             last_code = code
-            last_opts = ropts
-            blob += payload
             # 4.xx / 5.xx responses don't carry Block2 continuation —
             # bail with whatever we got. Caller decides if 4.xx is fatal.
             if code >> 5 != 2:
-                return code, blob
+                return code, blob, blocks, tok
+
+            # RFC 7959 §2.4: compare ETags across blocks, or we splice
+            # two versions of the resource into one buffer.
+            block_etag = next((v for n, v in ropts if n == ETAG), None)
+            if num == 0:
+                etag = block_etag
+            elif etag is not None and block_etag != etag:
+                raise _EtagChanged()
+
+            blob += payload
             b2 = [v for n, v in ropts if n == BLOCK2]
-            more = 0
-            if b2:
-                bv = int.from_bytes(b2[0], 'big')
-                more = (bv >> 3) & 1
-                server_szx = bv & 0x07
-                if server_szx != szx:
-                    szx = server_szx
+            if not b2:
+                break
+            _, more, server_szx = block_fields(b2[0])
             if not more:
                 break
-            num += 1
+            if server_szx != szx:
+                # Server negotiated the block size down. Block numbers
+                # are indices into the new size, so the next one has to
+                # come off the byte offset we have actually accumulated,
+                # not off num + 1.
+                szx = server_szx
+                num = len(blob) >> (szx + 4)
+            else:
+                num += 1
             if num > self.MAX_BLOCKS:
                 raise BlockwiseError()
-        return last_code, blob
+        return last_code, blob, blocks, tok
+
+    def _exchange_block(self, tok, path_segs, query, num, szx, deadline):
+        """Send one block request under `tok` and return its response
+        container, retransmitting up to _BLOCK_MAX_ATTEMPTS times.
+
+        A response whose Block2 NUM is not the one we asked for is a
+        retransmit of an earlier block, not the next one. Concatenating
+        it would corrupt the buffer, so keep waiting on the same
+        attempt budget instead."""
+        for attempt in range(_BLOCK_MAX_ATTEMPTS):
+            ev = threading.Event()
+            container = {}
+            with self._state_lock:
+                self._pending[tok] = (ev, container)
+            try:
+                mid = self._next_mid()
+                opts = [(URI_PATH, s.encode()) for s in path_segs]
+                for q in query:
+                    opts.append((URI_QUERY, q.encode()))
+                opts.append((ACCEPT, CF_CBOR))
+                if num > 0:
+                    opts.append((BLOCK2, block_value(num, 0, szx)))
+                self._send_dgram(
+                    build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
+                while True:
+                    per_wait = min(_BLOCK_ACK_TIMEOUT,
+                                   max(0.1, deadline - time.time()))
+                    if not self._wait_for_block(ev, per_wait):
+                        break  # attempt timed out
+                    if 'err' in container or self._block_num_matches(
+                            container, num):
+                        return container
+                    logger.debug(
+                        "GET %s /%s block %d: stale block, still waiting",
+                        self.host, '/'.join(path_segs), num)
+                    ev.clear()
+                    container.clear()
+                    if deadline - time.time() <= 0:
+                        break
+                remaining = deadline - time.time()
+                if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1:
+                    logger.debug(
+                        "GET %s /%s block %d: timed out after %d attempt(s)",
+                        self.host, '/'.join(path_segs), num, attempt + 1,
+                    )
+                    raise SessionTimeoutError()
+                logger.debug(
+                    "GET %s /%s block %d: attempt %d/%d timeout, retrying",
+                    self.host, '/'.join(path_segs), num,
+                    attempt + 1, _BLOCK_MAX_ATTEMPTS,
+                )
+            finally:
+                with self._state_lock:
+                    self._pending.pop(tok, None)
+        raise SessionTimeoutError()
+
+    def _wait_for_block(self, ev, per_wait):
+        """Wait for one block response, giving up early if the reader
+        dies underneath us.
+
+        Only the reader thread can resolve a token, so once it is gone
+        the wait can never succeed. Polling in slices turns what would
+        be a full per-block timeout into an immediate SessionClosedError,
+        which is the same fail-fast contract get() gets from _check_live()
+        at entry — it just has to hold for every block, not only the
+        first."""
+        deadline = time.time() + per_wait
+        while True:
+            slice_s = min(_BLOCK_LIVENESS_POLL_S, deadline - time.time())
+            if slice_s <= 0:
+                return False
+            if ev.wait(slice_s):
+                return True
+            self._check_live()
+
+    @staticmethod
+    def _block_num_matches(container, num):
+        """True if this response carries the block we asked for. A
+        response with no Block2 option is the whole representation, so
+        it only answers block 0."""
+        if container.get('code', 0) >> 5 != 2:
+            return True     # error responses end the transfer either way
+        b2 = [v for n, v in container.get('options', ()) if n == BLOCK2]
+        if not b2:
+            return num == 0
+        return block_fields(b2[0])[0] == num
 
     def post(self, path_segs, body_cbor, timeout=8.0):
         """Single-frame POST with a CBOR-encoded body. Returns
@@ -738,7 +1020,8 @@ class DtlsCoapSession:
                               body_cbor)
         ev = threading.Event()
         container = {}
-        self._pending[tok] = (ev, container)
+        with self._state_lock:
+            self._pending[tok] = (ev, container)
         try:
             self._send_dgram(datagram)
             if not ev.wait(timeout):
@@ -747,7 +1030,8 @@ class DtlsCoapSession:
                 raise container['err']
             return container['code'], container['payload']
         finally:
-            self._pending.pop(tok, None)
+            with self._state_lock:
+                self._pending.pop(tok, None)
 
     def ping(self):
         """RFC 7252 §4.4 CoAP Ping — empty CON, no token, no payload.

tests/test_coap_wire.py

@@ -2,7 +2,8 @@ import pytest
 
 from smartthings_local.errors import MalformedMessageError
 from smartthings_local.protocol.coap import (
-    build_coap, parse_coap, encode_options, block_value, fmt_code,
+    build_coap, parse_coap, encode_options, block_value, block_fields,
+    fmt_code,
     TYPE_CON, METHOD_GET, URI_PATH, ACCEPT, CF_CBOR, BLOCK2,
 )
 
@@ -43,6 +44,23 @@ def test_block_value_promotes_to_two_bytes_when_num_is_large():
     assert len(v) == 2
 
 
+@pytest.mark.parametrize('num, more, szx', [
+    (0, 0, 0),
+    (0, 1, 6),
+    (2, 1, 6),
+    (1, 0, 4),
+    (0xFFF, 0, 0),
+    (0xFFFFF, 1, 7),
+])
+def test_block_fields_inverts_block_value(num, more, szx):
+    assert block_fields(block_value(num, more, szx)) == (num, more, szx)
+
+
+def test_block_fields_treats_empty_value_as_block_zero():
+    # RFC 7959 §2.2: a zero-length Block option means num=0, m=0, szx=0.
+    assert block_fields(b'') == (0, 0, 0)
+
+
 def test_fmt_code_formats_class_dot_detail():
     assert fmt_code(0x45) == '2.05'
     assert fmt_code(0x84) == '4.04'

tests/test_observe_block2.py

@@ -0,0 +1,508 @@
+"""Blockwise OBSERVE notifications (QuiteYellow/SmartThings-Local#39).
+
+A notification carries only the first block of the representation
+(RFC 7959 §2.6). Before this, _dispatch_coap handed that first block
+straight to on_notification and the consumer decoded a truncated CBOR
+buffer. These tests pin the replacement: a truncated notification is
+withheld, the resource is re-read from block 0 under a fresh one-shot
+token, and only the reassembled representation reaches the callback.
+
+The re-read starts at block 0 rather than continuing at NUM=1 for two
+reasons, both recorded on #39: RFC 7959 §3.4 forbids continuing on the
+observation's token, and Samsung's RT-OCF drops a transfer that opens
+at NUM>0 under a token it has not seen.
+"""
+import logging
+import socket
+import threading
+import time
+
+import pytest
+from OpenSSL import SSL
+
+from smartthings_local.errors import BlockwiseError
+from smartthings_local.protocol import dtls_session
+from smartthings_local.protocol.coap import (
+    BLOCK2, ETAG, METHOD_GET, OBSERVE, TYPE_ACK, TYPE_NON,
+    block_fields, block_value, build_coap, parse_coap,
+)
+from smartthings_local.protocol.dtls_session import DtlsCoapSession
+
+SZX = 6  # 1024-byte blocks, the only size these appliances honour
+_LOGGER_NAME = "smartthings_local.protocol.dtls_session"
+
+
+class _NullAuth:
+    """Structural AuthenticationProvider — never configured, we skip connect()."""
+
+    def configure_context(self, _context):
+        return None
+
+
+class _LoopbackConn:
+    """SSL.Connection stand-in that answers requests from a script.
+
+    `responder` is called with each parsed request and returns a list of
+    CoAP datagrams to hand back (possibly empty, to model a silent
+    device). Responses surface on recv() the way decrypted records do.
+    """
+
+    def __init__(self, responder):
+        self._responder = responder
+        self._inbox = []
+        self._lock = threading.Lock()
+        self.sent = []
+
+    # -- client -> device
+    def send(self, datagram):
+        self.sent.append(parse_coap(datagram))
+        for reply in self._responder(parse_coap(datagram)):
+            with self._lock:
+                self._inbox.append(reply)
+        return len(datagram)
+
+    def bio_read(self, _n):
+        return b""
+
+    # -- device -> client
+    def inject(self, datagram):
+        """Push a device-initiated frame (an OBSERVE notification)."""
+        with self._lock:
+            self._inbox.append(datagram)
+
+    def bio_write(self, _datagram):
+        return None
+
+    def recv(self, _n):
+        with self._lock:
+            if self._inbox:
+                return self._inbox.pop(0)
+        raise SSL.WantReadError()
+
+    def shutdown(self):
+        return None
+
+    def pending(self):
+        with self._lock:
+            return len(self._inbox)
+
+
+class _PumpSock:
+    """UDP socket stand-in. recv() returns a dummy datagram whenever the
+    connection has something decrypted waiting, so the reader loop keeps
+    pumping; otherwise it times out like a real socket."""
+
+    def __init__(self, conn):
+        self._conn = conn
+        self.closed = False
+
+    def settimeout(self, _value):
+        return None
+
+    def recv(self, _n):
+        for _ in range(20):
+            if self.closed:
+                raise OSError("closed")
+            if self._conn.pending():
+                return b"\x00"
+            time.sleep(0.005)
+        raise socket.timeout()
+
+    def send(self, data):
+        return len(data)
+
+    def close(self):
+        self.closed = True
+
+
+def _make_session(responder, **kwargs):
+    calls = []
+    sess = DtlsCoapSession(
+        "host", 1234, auth=_NullAuth(),
+        on_notification=lambda href, payload: calls.append((href, payload)),
+        **kwargs)
+    sess.conn = _LoopbackConn(responder)
+    sess.sock = _PumpSock(sess.conn)
+    sess.start_reader()
+    return sess, calls
+
+
+def _notification(tok, payload, *, block2=None, mtype=TYPE_NON, obs=1):
+    opts = [(OBSERVE, bytes([obs]))]
+    if block2 is not None:
+        opts.append((BLOCK2, block2))
+    return build_coap(mtype, 0x45, 0x1234, tok, opts, payload)
+
+
+def _content(tok, mid, payload, *, block2=None, etag=None):
+    opts = []
+    if etag is not None:
+        opts.append((ETAG, etag))
+    if block2 is not None:
+        opts.append((BLOCK2, block2))
+    return build_coap(TYPE_ACK, 0x45, mid, tok, opts, payload)
+
+
+def _requested_block(request):
+    """(num, szx) the request asked for, or (0, None) with no Block2."""
+    _, _, _, _, opts, _ = request
+    b2 = [v for n, v in opts if n == BLOCK2]
+    if not b2:
+        return 0, None
+    num, _, szx = block_fields(b2[0])
+    return num, szx
+
+
+def _wait_for(predicate, timeout=3.0):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return False
+
+
+def _close(sess):
+    sess.close()
+    sess.join()
+
+
+# --------------------------------------------------------------------
+# The no-regression case
+
+
+def test_single_block_notification_is_delivered_inline():
+    sess, calls = _make_session(lambda request: [])
+    try:
+        tok = sess.subscribe(["oven", "vs", "0"])
+        sess.conn.inject(_notification(tok, b"\xa1\x01\x02"))
+
+        assert _wait_for(lambda: calls)
+        assert calls == [("/oven/vs/0", b"\xa1\x01\x02")]
+        # The subscribe GET is the only thing we sent — no refetch.
+        assert len(sess.conn.sent) == 1
+    finally:
+        _close(sess)
+
+
+def test_complete_block_zero_notification_is_delivered_inline():
+    """Block2 present but M=0 and NUM=0 means the whole representation
+    fit in one block. Nothing to fetch back."""
+    sess, calls = _make_session(lambda request: [])
+    try:
+        tok = sess.subscribe(["oven", "vs", "0"])
+        sess.conn.inject(
+            _notification(tok, b"\xa1\x01\x02",
+                          block2=block_value(0, 0, SZX)))
+
+        assert _wait_for(lambda: calls)
+        assert calls == [("/oven/vs/0", b"\xa1\x01\x02")]
+        assert len(sess.conn.sent) == 1
+    finally:
+        _close(sess)
+
+
+# --------------------------------------------------------------------
+# The fix
+
+
+def test_truncated_notification_is_refetched_and_reassembled():
+    blocks = [b"A" * 1024, b"B" * 40]
+
+    def responder(request):
+        _mtype, code, mid, tok, opts, _ = request
+        if code != METHOD_GET or any(n == OBSERVE for n, _ in opts):
+            return []           # the subscribe registration itself
+        num, _ = _requested_block(request)
+        more = 1 if num + 1 < len(blocks) else 0
+        return [_content(tok, mid, blocks[num],
+                         block2=block_value(num, more, SZX))]
+
+    sess, calls = _make_session(responder)
+    try:
+        tok = sess.subscribe(["mode", "vs", "0"])
+        sess.conn.inject(
+            _notification(tok, blocks[0],
+                          block2=block_value(0, 1, SZX)))
+
+        assert _wait_for(lambda: calls), "callback never fired"
+        assert calls == [("/mode/vs/0", b"".join(blocks))]
+    finally:
+        _close(sess)
+
+
+def test_refetch_uses_a_fresh_one_shot_token_not_the_observe_token():
+    """RFC 7959 §3.4: the requests for additional blocks cannot use the
+    token of the observation relationship."""
+    blocks = [b"A" * 1024, b"B" * 40]
+
+    def responder(request):
+        _mtype, _code, mid, tok, opts, _ = request
+        if any(n == OBSERVE for n, _ in opts):
+            return []
+        num, _ = _requested_block(request)
+        more = 1 if num + 1 < len(blocks) else 0
+        return [_content(tok, mid, blocks[num],
+                         block2=block_value(num, more, SZX))]
+
+    sess, calls = _make_session(responder)
+    try:
+        observe_tok = sess.subscribe(["mode", "vs", "0"])
+        assert len(observe_tok) == 1, "OBSERVE registrations use 1-byte tokens"
+        sess.conn.inject(
+            _notification(observe_tok, blocks[0],
+                          block2=block_value(0, 1, SZX)))
+        assert _wait_for(lambda: calls)
+
+        refetch = [r for r in sess.conn.sent
+                   if not any(n == OBSERVE for n, _ in r[4])]
+        assert refetch, "no refetch request was sent"
+        tokens = {r[3] for r in refetch}
+        assert observe_tok not in tokens
+        assert all(len(t) == 4 for t in tokens), "one-shot tokens are 4-byte"
+        assert len(tokens) == 1, "the transfer must hold one token throughout"
+
+        # And the transfer restarts at block 0 rather than continuing at 1.
+        assert _requested_block(refetch[0])[0] == 0
+        assert [_requested_block(r)[0] for r in refetch] == [0, 1]
+        # No Observe option on a continuation request.
+        assert not any(n == OBSERVE for r in refetch for n, _ in r[4])
+    finally:
+        _close(sess)
+
+
+def test_silent_device_drops_the_notification_without_delivering_a_partial():
+    sess, calls = _make_session(lambda request: [])
+    try:
+        tok = sess.subscribe(["mode", "vs", "0"])
+        sess.conn.inject(
+            _notification(tok, b"A" * 1024,
+                          block2=block_value(0, 1, SZX)))
+        # Give the worker a chance to try and fail. _BLOCK_ACK_TIMEOUT is
+        # 4s per attempt, so we only need to see that nothing partial got
+        # through in the meantime.
+        assert not _wait_for(lambda: calls, timeout=0.6)
+        assert sess._reader_thread.is_alive(), "reader must survive"
+    finally:
+        _close(sess)
+
+
+def test_non_2xx_refetch_is_dropped_rather_than_delivered():
+    def responder(request):
+        _mtype, _code, mid, tok, opts, _ = request
+        if any(n == OBSERVE for n, _ in opts):
+            return []
+        return [build_coap(TYPE_ACK, 0x84, mid, tok, [], b"")]
+
+    sess, calls = _make_session(responder)
+    try:
+        tok = sess.subscribe(["mode", "vs", "0"])
+        sess.conn.inject(
+            _notification(tok, b"A" * 1024,
+                          block2=block_value(0, 1, SZX)))
+        assert not _wait_for(lambda: calls, timeout=0.6)
+    finally:
+        _close(sess)
+
+
+def test_notification_burst_collapses_to_one_refetch_per_resource():
+    blocks = [b"A" * 1024, b"B" * 40]
+    gate = threading.Event()
+
+    def responder(request):
+        _mtype, _code, mid, tok, opts, _ = request
+        if any(n == OBSERVE for n, _ in opts):
+            return []
+        gate.wait(2.0)          # hold the first transfer open
+        num, _ = _requested_block(request)
+        more = 1 if num + 1 < len(blocks) else 0
+        return [_content(tok, mid, blocks[num],
+                         block2=block_value(num, more, SZX))]
+
+    sess, calls = _make_session(responder)
+    try:
+        tok = sess.subscribe(["mode", "vs", "0"])
+        for seq in range(5):
+            sess.conn.inject(
+                _notification(tok, blocks[0], obs=seq + 1,
+                              block2=block_value(0, 1, SZX)))
+        # Five notifications, one queue entry: latest wins per href.
+        assert _wait_for(lambda: sess._refetch_pending or sess.conn.sent[1:])
+        assert len(sess._refetch_pending) <= 1
+        gate.set()
+
+        assert _wait_for(lambda: calls)
+        assert _wait_for(
+            lambda: not sess._refetch_pending and len(calls) >= 1)
+        time.sleep(0.2)
+        # Two transfers at most: the one in flight when the burst landed,
+        # plus one for the final state.
+        starts = [r for r in sess.conn.sent
+                  if not any(n == OBSERVE for n, _ in r[4])
+                  and _requested_block(r)[0] == 0]
+        assert len(starts) <= 2, f"{len(starts)} refetches for one burst"
+        assert calls[-1] == ("/mode/vs/0", b"".join(blocks))
+    finally:
+        gate.set()
+        _close(sess)
+
+
+def test_close_during_a_queued_refetch_stops_the_worker():
+    sess, _calls = _make_session(lambda request: [])
+    tok = sess.subscribe(["mode", "vs", "0"])
+    sess.conn.inject(
+        _notification(tok, b"A" * 1024, block2=block_value(0, 1, SZX)))
+    assert _wait_for(lambda: sess._refetch_thread is not None)
+
+    sess.close()
+    sess.join()     # hangs if the worker outlives the session
+    assert not sess._refetch_thread.is_alive()
+
+
+def test_refetch_worker_exits_when_the_reader_dies():
+    sess, _calls = _make_session(lambda request: [])
+    tok = sess.subscribe(["mode", "vs", "0"])
+    sess.conn.inject(
+        _notification(tok, b"A" * 1024, block2=block_value(0, 1, SZX)))
+    assert _wait_for(lambda: sess._refetch_thread is not None)
+
+    # Kill the reader the way a socket error does, without close().
+    sess.sock.closed = True
+    assert _wait_for(lambda: not sess._reader_running.is_set(), timeout=5.0)
+    sess._refetch_thread.join(6.0)
+    assert not sess._refetch_thread.is_alive()
+    sess.close()
+
+
+def test_debug_bridge_promotes_the_refetch_outcome_to_info(monkeypatch, caplog):
+    """The hardware validation for #39 reads this line to confirm which
+    token the re-read used, so it has to survive the bridge's INFO
+    default. Without DEBUG_BRIDGE it stays at debug."""
+    monkeypatch.setattr(dtls_session, "DEBUG_BRIDGE", True)
+    blocks = [b"A" * 1024, b"B" * 40]
+
+    def responder(request):
+        _mtype, _code, mid, tok, opts, _ = request
+        if any(n == OBSERVE for n, _ in opts):
+            return []
+        num, _ = _requested_block(request)
+        more = 1 if num + 1 < len(blocks) else 0
+        return [_content(tok, mid, blocks[num],
+                         block2=block_value(num, more, SZX))]
+
+    sess, calls = _make_session(responder)
+    try:
+        with caplog.at_level(logging.INFO, logger=_LOGGER_NAME):
+            tok = sess.subscribe(["mode", "vs", "0"])
+            sess.conn.inject(
+                _notification(tok, blocks[0], block2=block_value(0, 1, SZX)))
+            assert _wait_for(lambda: calls)
+
+        line = next((r.getMessage() for r in caplog.records
+                     if r.getMessage().startswith("refetch /mode/vs/0")), None)
+        assert line is not None, "no refetch line at INFO"
+        assert "blocks=2" in line
+        assert f"bytes={sum(len(b) for b in blocks)}" in line
+        assert line.endswith("ok")
+        # The token in the line is the one-shot token, not the observe one.
+        assert f"tok={tok.hex()} " not in line
+    finally:
+        _close(sess)
+
+
+# --------------------------------------------------------------------
+# Shared Block2 loop hardening
+
+
+def test_stale_block_number_is_not_concatenated():
+    """A retransmit of block 0 arriving while we wait for block 1 must
+    not be appended as if it were block 1."""
+    served = []
+
+    def responder(request):
+        _mtype, _code, mid, tok, _opts, _ = request
+        num, _ = _requested_block(request)
+        served.append(num)
+        if num == 0:
+            return [_content(tok, mid, b"A" * 1024,
+                             block2=block_value(0, 1, SZX))]
+        # Answer the block-1 request with a duplicate of block 0 first.
+        return [
+            _content(tok, mid, b"A" * 1024, block2=block_value(0, 1, SZX)),
+            _content(tok, mid, b"B" * 40, block2=block_value(1, 0, SZX)),
+        ]
+
+    sess, _calls = _make_session(responder)
+    try:
+        code, payload = sess.get(["mode", "vs", "0"], timeout=5.0)
+        assert code == 0x45
+        assert payload == b"A" * 1024 + b"B" * 40
+    finally:
+        _close(sess)
+
+
+def test_etag_change_mid_transfer_restarts_then_fails():
+    etags = [b"\x01", b"\x02", b"\x03", b"\x04"]
+
+    def responder(request):
+        _mtype, _code, mid, tok, _opts, _ = request
+        num, _ = _requested_block(request)
+        # A different ETag on every single response: the representation
+        # never settles, so reassembly can never be consistent.
+        etag = etags.pop(0) if etags else b"\xff"
+        payload = b"A" * 1024 if num == 0 else b"B" * 40
+        more = 1 if num == 0 else 0
+        return [_content(tok, mid, payload, etag=etag,
+                         block2=block_value(num, more, SZX))]
+
+    sess, _calls = _make_session(responder)
+    try:
+        with pytest.raises(BlockwiseError):
+            sess.get(["mode", "vs", "0"], timeout=5.0)
+    finally:
+        _close(sess)
+
+
+def test_stable_etag_across_blocks_reassembles():
+    def responder(request):
+        _mtype, _code, mid, tok, _opts, _ = request
+        num, _ = _requested_block(request)
+        payload = b"A" * 1024 if num == 0 else b"B" * 40
+        more = 1 if num == 0 else 0
+        return [_content(tok, mid, payload, etag=b"\x77",
+                         block2=block_value(num, more, SZX))]
+
+    sess, _calls = _make_session(responder)
+    try:
+        code, payload = sess.get(["mode", "vs", "0"], timeout=5.0)
+        assert code == 0x45
+        assert payload == b"A" * 1024 + b"B" * 40
+    finally:
+        _close(sess)
+
+
+def test_szx_downshift_asks_for_the_block_after_what_we_have():
+    """Server answers block 0 at SZX=6 (1024B) then drops to SZX=4
+    (256B). Block numbers index the new size, so the next request is
+    block 4, not block 1."""
+    requested = []
+
+    def responder(request):
+        _mtype, _code, mid, tok, _opts, _ = request
+        num, szx = _requested_block(request)
+        requested.append((num, szx))
+        if num == 0:
+            return [_content(tok, mid, b"A" * 1024,
+                             block2=block_value(0, 1, 4))]
+        return [_content(tok, mid, b"B" * 100,
+                         block2=block_value(num, 0, 4))]
+
+    sess, _calls = _make_session(responder)
+    try:
+        code, payload = sess.get(["mode", "vs", "0"], timeout=5.0)
+        assert code == 0x45
+        assert payload == b"A" * 1024 + b"B" * 100
+        # 1024 bytes in hand at 256B blocks = blocks 0..3 done, ask for 4.
+        assert requested[1] == (4, 4)
+    finally:
+        _close(sess)