From b977df589d313a16f083cffb39550550820b0cb8 Mon Sep 17 00:00:00 2001 From: Jack Nagy Date: Sun, 23 Aug 2026 18:58:47 +0100 Subject: [PATCH] feat(mqtt): expose write retransmission as WRITE_MAX_ATTEMPTS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #54 added write_max_attempts to DtlsCoapSession, off by default, on the argument that retransmitting into an appliance already dropping under load turns one lost write into several and that RFC 7252 §4.5 dedupe is unverified on RT-OCF. Turning it on is meant to follow a measurement. The flag was constructor-only and session_once() did not pass it, so the reference bridge was the one deployment that could take that measurement and the only one that could not reach the switch. WRITE_MAX_ATTEMPTS now sits in SharedConfig beside the other int knobs and reaches the session. It defaults to 1, which is a single send and today's behaviour, so a bridge that never sets it is unchanged. docker-compose passes .env through with env_file, so this needs no compose edit on the host. Tests cover the default, the env var, a non-numeric value stopping the process at startup the way the other int knobs do, and the call site naming the SharedConfig field so the two stay wired together. --- mqtt_demo/.env.example | 12 +++++ mqtt_demo/bridge.py | 1 + mqtt_demo/config.py | 2 + tests/test_bridge_write_attempts.py | 83 +++++++++++++++++++++++++++++ 4 files changed, 98 insertions(+) create mode 100644 tests/test_bridge_write_attempts.py diff --git a/mqtt_demo/.env.example b/mqtt_demo/.env.example index 9895d1b..4cca93e 100644 --- a/mqtt_demo/.env.example +++ b/mqtt_demo/.env.example @@ -53,6 +53,18 @@ HA_DISCOVERY_PREFIX=homeassistant HEALTH_INTERVAL_S=60 PING_INTERVAL_S=25 +# Write retransmission. A CoAP read retransmits each block; a write is +# sent once, so one lost datagram is an unrecoverable command while a +# lost read recovers silently. Raising this lets post() resend the +# byte-identical CON inside the caller's own timeout, so a device +# implementing RFC 7252 §4.5 answers the duplicate from its dedupe cache +# instead of running the write twice. +# Leave it at 1 (send once, today's behaviour) unless a lost write has +# been measured. Retransmitting into an appliance that is already +# dropping under load turns one lost write into several, and §4.5 dedupe +# is unverified on Samsung's RT-OCF. +WRITE_MAX_ATTEMPTS=1 + # Container TZ. TZ=Europe/London diff --git a/mqtt_demo/bridge.py b/mqtt_demo/bridge.py index 9666c24..f15c554 100644 --- a/mqtt_demo/bridge.py +++ b/mqtt_demo/bridge.py @@ -348,6 +348,7 @@ def session_once(self): key_path=self.shared.KEY_PATH, on_notification=self._on_notification, local_port=DTLS_LOCAL_PORT_BASE + self.app.index, + write_max_attempts=self.shared.WRITE_MAX_ATTEMPTS, ) sess.connect() self.port = port diff --git a/mqtt_demo/config.py b/mqtt_demo/config.py index 3da393a..53b4320 100644 --- a/mqtt_demo/config.py +++ b/mqtt_demo/config.py @@ -58,6 +58,7 @@ class SharedConfig: HA_DISCOVERY_PREFIX: str HEALTH_INTERVAL_S: int PING_INTERVAL_S: int + WRITE_MAX_ATTEMPTS: int @classmethod def from_env(cls) -> 'SharedConfig': @@ -72,6 +73,7 @@ def from_env(cls) -> 'SharedConfig': 'homeassistant'), HEALTH_INTERVAL_S=int(os.getenv('HEALTH_INTERVAL_S', '60')), PING_INTERVAL_S=int(os.getenv('PING_INTERVAL_S', '25')), + WRITE_MAX_ATTEMPTS=int(os.getenv('WRITE_MAX_ATTEMPTS', '1')), ) diff --git a/tests/test_bridge_write_attempts.py b/tests/test_bridge_write_attempts.py new file mode 100644 index 0000000..d6a8cb3 --- /dev/null +++ b/tests/test_bridge_write_attempts.py @@ -0,0 +1,83 @@ +"""WRITE_MAX_ATTEMPTS plumbing: env var -> SharedConfig -> the session. + +The library flag exists so a lost write can be measured on real hardware +before retransmission is turned on anywhere. That measurement happens on +the bridge, so the bridge has to be able to set it -- and, far more +importantly, has to keep defaulting to one send until someone chooses +otherwise. +""" +import inspect +import logging +import types + +import pytest + +import mqtt_demo.bridge as bridge +from mqtt_demo.config import SharedConfig + + +@pytest.fixture(autouse=True) +def _clear_env(monkeypatch): + monkeypatch.delenv('WRITE_MAX_ATTEMPTS', raising=False) + + +def test_default_is_one_send(): + # The default has to stay today's behaviour: retransmitting into an + # appliance already dropping under load turns one lost write into + # several, and §4.5 dedupe is unverified on RT-OCF. + assert SharedConfig.from_env().WRITE_MAX_ATTEMPTS == 1 + + +def test_env_var_is_read(monkeypatch): + monkeypatch.setenv('WRITE_MAX_ATTEMPTS', '3') + assert SharedConfig.from_env().WRITE_MAX_ATTEMPTS == 3 + + +def test_a_non_numeric_value_fails_at_startup(monkeypatch): + # Same contract as the other int knobs: a typo stops the process + # rather than silently reverting to a default nobody asked for. + monkeypatch.setenv('WRITE_MAX_ATTEMPTS', 'yes') + with pytest.raises(ValueError): + SharedConfig.from_env() + + +class _StopAfterConstruction(Exception): + """Ends session_once at the point this test cares about, before it + starts a reader thread and a supervision loop.""" + + +def test_session_once_passes_it_through(monkeypatch): + """The plumbing that makes the flag reachable at all: without this + the constructor argument exists but nothing on the bridge sets it.""" + captured = {} + + class _FakeSession: + def __init__(self, host, port, **kw): + captured.update(kw) + + def connect(self): + raise _StopAfterConstruction() + + monkeypatch.setattr(bridge, 'DtlsCoapSession', _FakeSession) + + b = bridge.PushBridge.__new__(bridge.PushBridge) + b.app = types.SimpleNamespace(ip='192.0.2.9', ocf_port=49155, index=0) + b.shared = types.SimpleNamespace( + CERT_PATH='/tmp/cert.pem', KEY_PATH='/tmp/key.pem', + WRITE_MAX_ATTEMPTS=4, + ) + b.log = logging.getLogger('test-bridge') + b._on_notification = lambda *a: None + b._resolve_port = lambda: 49155 # no DTLS probe against a real host + + with pytest.raises(_StopAfterConstruction): + bridge.PushBridge.session_once(b) + + assert captured['write_max_attempts'] == 4 + + +def test_shared_config_field_is_wired_into_the_session_call(): + """A guard against the field being added and then never read: the + call site must name the SharedConfig attribute, not a literal.""" + source = inspect.getsource(bridge.PushBridge.session_once) + assert 'write_max_attempts=self.shared.WRITE_MAX_ATTEMPTS' in source