bhetherman/localthings
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: CoAP inter-request pacing to prevent RT-OCF rate limit drops

Browse Source 
Samsung appliances silently drop CoAP requests when hit faster than their
firmware ceiling (~8–14 req/s measured; dishwasher unknown). Multi-block
Block2 fetches on LAN hit ~100 req/s instantaneously, causing the observed
burst timeouts on block 4 and block 10 of /device/0.

Adds rate_limit_rps (default 5 req/s / 200 ms) to DtlsCoapSession with a
pace() method that enforces the inter-block delay via _stop.wait() so session
teardown interrupts the sleep cleanly. Coordinator calls pace() between hrefs
in _poll_hrefs_blocking. Also adds device IP to all GET log lines for easier
per-device filtering.
This commit is contained in:
Marc Billow
2026-06-30 22:40:14 -05:00
parent 9ea4ddd3a3
commit 022a192bca
7 changed files with 130 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/const.py
+1 -2
View File
@@ -11,5 +11,4 @@ CONF_LEAF_KEY_PEM = "leaf_key_pem"
PROBE_PORTS = [49154, 49155]
ACTIVE_INTERVAL_S = 5.0
IDLE_INTERVAL_S = 30.0
SUMMARY_INTERVAL_S = 30.0
custom_components/localthings/coordinator.py
+98 -19
View File
@@ -18,12 +18,12 @@ from .ocf.coap_dtls import DtlsCoapSession
from .ocf.registry.by_type import for_device
from .ocf.registry.discovery import discover, BoundEntity
from .ocf.registry import CAPABILITIES
from .ocf.registry.adapter import flatten, is_active
from .ocf.registry.adapter import flatten
from .ocf.registry.identity import read_identity, DeviceIdentity
from .const import (
DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM,
ACTIVE_INTERVAL_S, IDLE_INTERVAL_S,
SUMMARY_INTERVAL_S,
)
_LOGGER = logging.getLogger(__name__)
@@ -43,7 +43,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
hass,
_LOGGER,
name=f"{DOMAIN}_{entry.data[CONF_HOST]}",
update_interval=timedelta(seconds=IDLE_INTERVAL_S),
update_interval=timedelta(seconds=SUMMARY_INTERVAL_S),
)
self._entry = entry
self._session: DtlsCoapSession | None = None
@@ -57,6 +57,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
name=f"Samsung Appliance ({entry.data[CONF_HOST]})",
manufacturer="Samsung",
)
self._session_lock = asyncio.Lock()
self._subpoll_task: asyncio.Task | None = None
self._hot_hrefs: list[str] = []
self._warm_hrefs: list[str] = []
# ------------------------------------------------------------------
# Session management (all blocking — must run in executor)
@@ -93,6 +97,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
pass
async def async_close(self) -> None:
if self._subpoll_task is not None:
self._subpoll_task.cancel()
self._subpoll_task = None
await self.hass.async_add_executor_job(self._close_session)
def _poll_once(self) -> dict[str, dict]:
@@ -114,6 +121,59 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
raise RuntimeError(f"poll cbor decode: {e}") from e
return parse_device0_batch(body) if isinstance(body, list) else {}
def _poll_hrefs_blocking(self, hrefs: list[str]) -> dict[str, dict]:
"""GET individual hrefs sequentially. Does not reconnect on failure. Blocking."""
if self._session is None:
return {}
results = {}
first = True
for href in hrefs:
if not first:
self._session.pace()
first = False
try:
path = [s for s in href.strip('/').split('/')]
code, payload = self._session.get(path, timeout=10.0)
if code == 0x45 and payload:
rep = cbor2.loads(payload)
if isinstance(rep, dict):
results[href] = rep
except Exception as e:
_LOGGER.debug("sub-poll %s: %s", href, e)
return results
# ------------------------------------------------------------------
# Sub-poll loop (runs between summary polls)
# ------------------------------------------------------------------
async def _run_subpolls(self) -> None:
"""Poll hot/warm hrefs in the gaps between summary polls.
Schedule over a SUMMARY_INTERVAL_S window:
hot hrefs every slot (10x, ~3 s apart)
warm hrefs every other slot (5x, ~6 s apart)
The lock prevents overlap with the summary poll or a concurrent sub-poll.
"""
hot = self._hot_hrefs
warm = self._warm_hrefs
if not hot and not warm:
return
step = SUMMARY_INTERVAL_S / 10 # 3.0 s
for i in range(1, 10): # slots 1..9 (T+3 s … T+27 s)
await asyncio.sleep(step)
hrefs = list(hot) + (list(warm) if i % 2 == 0 else [])
async with self._session_lock:
try:
updates = await self.hass.async_add_executor_job(
self._poll_hrefs_blocking, hrefs
)
except Exception as e:
_LOGGER.debug("sub-poll batch failed: %s", e)
continue
if updates:
self._last_resources.update(updates)
self.async_set_updated_data(flatten(self.bound, self._last_resources))
# ------------------------------------------------------------------
# Discovery (runs once on first successful poll)
# ------------------------------------------------------------------
@@ -148,37 +208,56 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
manufacturer=mfr,
model=model,
)
hot, warm = set(), set()
for be in bound:
tier = be.capability.poll_tier
if tier == 'hot':
hot.add(be.href)
elif tier == 'warm':
warm.add(be.href)
self._hot_hrefs = sorted(hot)
self._warm_hrefs = sorted(warm)
self._discovered = True
_LOGGER.info("discovered %d entities (serial=%s)", len(bound), serial)
_LOGGER.info(
"discovered %d entities (serial=%s) hot=%s warm=%s",
len(bound), serial, self._hot_hrefs, self._warm_hrefs,
)
# ------------------------------------------------------------------
# DataUpdateCoordinator hook
# ------------------------------------------------------------------
async def _async_update_data(self) -> dict[str, Any]:
try:
resources = await self.hass.async_add_executor_job(self._poll_once)
except Exception as e:
# One reconnect attempt — pause briefly so the device can
# clean up its DTLS session state before we knock again.
_LOGGER.warning("poll failed, reconnecting: %s", e)
await self.hass.async_add_executor_job(self._close_session)
await asyncio.sleep(5.0)
# Stop any in-flight sub-poll before taking the session lock.
if self._subpoll_task is not None:
self._subpoll_task.cancel()
self._subpoll_task = None
async with self._session_lock:
try:
resources = await self.hass.async_add_executor_job(self._poll_once)
except Exception as e2:
raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2
except Exception as e:
# One reconnect attempt — pause briefly so the device can
# clean up its DTLS session state before we knock again.
_LOGGER.warning("poll failed, reconnecting: %s", e)
await self.hass.async_add_executor_job(self._close_session)
await asyncio.sleep(5.0)
try:
resources = await self.hass.async_add_executor_job(self._poll_once)
except Exception as e2:
raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2
self._last_resources = resources
if not self._discovered:
self._run_discovery(resources)
# Adjust polling interval dynamically
active = is_active(self.bound, resources)
self.update_interval = timedelta(
seconds=ACTIVE_INTERVAL_S if active else IDLE_INTERVAL_S
)
# Schedule sub-polls for hot/warm hrefs between summary polls.
if self._hot_hrefs or self._warm_hrefs:
self._subpoll_task = self.hass.async_create_task(
self._run_subpolls(), name="localthings_subpoll"
)
return flatten(self.bound, resources)
custom_components/localthings/ocf/coap_dtls.py
+24 -9
View File
@@ -102,6 +102,13 @@ BLOCK_SZX = 6
_BLOCK_MAX_ATTEMPTS = 3
_BLOCK_ACK_TIMEOUT = 4.0
# 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
# (200 ms) is conservative enough for all tested devices; tune per device
# once the ceiling is measured empirically.
_DEFAULT_RATE_LIMIT_RPS = 5.0
def _vlen(v):
"""Variable-length integer encoder used in option deltas + lengths."""
@@ -222,13 +229,15 @@ class DtlsCoapSession:
MAX_BLOCKS = 32 # safety bound for Block2 fetches
def __init__(self, host, port, cert_pem, key_pem,
on_notification=None, mtu=1200):
on_notification=None, mtu=1200,
rate_limit_rps: float = _DEFAULT_RATE_LIMIT_RPS):
self.host = host
self.port = port
self.cert_pem = cert_pem
self.key_pem = key_pem
self.on_notification = on_notification # fn(href, payload_bytes)
self.mtu = mtu
self._min_req_interval = 1.0 / rate_limit_rps
self.sock = None
self.conn = None
@@ -254,6 +263,10 @@ class DtlsCoapSession:
self._stop = threading.Event()
self._reader_thread = None
def pace(self) -> None:
"""Sleep one rate-limit interval. Uses _stop so session teardown wakes it."""
self._stop.wait(self._min_req_interval)
# ---- lifecycle ---------------------------------------------------
def connect(self):
@@ -566,8 +579,8 @@ class DtlsCoapSession:
raise ConnectionError("DTLS session closed")
tok = self._next_tok()
logger.debug(
"GET /%s token=%s timeout=%.1fs",
'/'.join(path_segs), tok.hex(), timeout,
"GET %s /%s token=%s timeout=%.1fs",
self.host, '/'.join(path_segs), tok.hex(), timeout,
)
blob = b''
num = 0
@@ -576,6 +589,8 @@ class DtlsCoapSession:
last_opts = []
deadline = time.time() + timeout
while True:
if num > 0:
self.pace()
container = {}
for attempt in range(_BLOCK_MAX_ATTEMPTS):
ev = threading.Event()
@@ -598,14 +613,14 @@ class DtlsCoapSession:
remaining = deadline - time.time()
if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1:
logger.debug(
"GET /%s block %d: timed out after %d attempt(s)",
'/'.join(path_segs), num, attempt + 1,
"GET %s /%s block %d: timed out after %d attempt(s)",
self.host, '/'.join(path_segs), num, attempt + 1,
)
raise TimeoutError(
f"GET /{'/'.join(path_segs)} block {num} timeout")
logger.debug(
"GET /%s block %d: attempt %d/%d timeout, retrying",
'/'.join(path_segs), num,
"GET %s /%s block %d: attempt %d/%d timeout, retrying",
self.host, '/'.join(path_segs), num,
attempt + 1, _BLOCK_MAX_ATTEMPTS,
)
finally:
@@ -620,8 +635,8 @@ class DtlsCoapSession:
)
else:
logger.debug(
"GET /%s block %d: %s (%d bytes)",
'/'.join(path_segs), num,
"GET %s /%s block %d: %s (%d bytes)",
self.host, '/'.join(path_segs), num,
fmt_code(code_seen),
len(container.get('payload', b'')),
)
custom_components/localthings/ocf/registry/capabilities/common.py
+1
View File
@@ -73,6 +73,7 @@ POWER = Capability(
ALARMS = Capability(
href='/alarms/vs/0',
poll_tier='hot',
entities=(
SensorDesc(key='alarm_code', field='x.com.samsung.da.items',
name='Alarm code', icon='mdi:alert',
custom_components/localthings/ocf/registry/capability.py
+1 -1
View File
@@ -11,7 +11,7 @@ from .entities import SamsungEntityDescription
class Capability:
href: Optional[str] = None
entities: tuple[SamsungEntityDescription, ...] = ()
poll_tier: str = 'warm' # 'hot' | 'warm' | 'cold'
poll_tier: str = 'cold' # 'hot' | 'warm' | 'cold'
rt_filter: Optional[str] = None # bind only if rt_filter in rep.get('rt', ())
href_prefix: Optional[str] = None # pattern caps only: bind only if href starts with this
strip_prefix_in_key: bool = False # strip href_prefix segs before building key_override
tests/localthings/test_coordinator.py
+4 -4
View File
@@ -8,7 +8,7 @@ import pytest
from homeassistant.core import HomeAssistant
from custom_components.localthings.const import (
ACTIVE_INTERVAL_S, DOMAIN, IDLE_INTERVAL_S,
DOMAIN, SUMMARY_INTERVAL_S,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
@@ -42,15 +42,15 @@ async def test_device_info_populated(
)
async def test_idle_interval_by_default(
async def test_summary_interval(
hass: HomeAssistant, mock_entry, mock_coordinator_session
) -> None:
"""Interval defaults to idle when no active_when fires."""
"""Summary poll interval is always SUMMARY_INTERVAL_S."""
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
assert coordinator.update_interval == timedelta(seconds=IDLE_INTERVAL_S)
assert coordinator.update_interval == timedelta(seconds=SUMMARY_INTERVAL_S)
async def test_update_failed_on_persistent_poll_error(
tests/test_capability.py
+1 -1
View File
@@ -5,7 +5,7 @@ from custom_components.localthings.ocf.registry.entities import BinarySensorDesc
def test_capability_defaults():
c = Capability(href='/kidslock/vs/0',
entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),))
assert c.poll_tier == 'warm'
assert c.poll_tier == 'cold'
assert c.active_when is None
assert len(c.entities) == 1