From f1c22ed159ac89342baf21c89906b377dfc423df Mon Sep 17 00:00:00 2001 From: emmanuel Date: Sat, 4 Jul 2026 23:16:18 -0700 Subject: [PATCH 1/2] fix: MQTT threading model and event emitter correctness - Marshal MQTT message parse/dispatch from the AWS CRT network thread onto the asyncio event loop; keeps the SDK callback trivial, prevents blocking user callbacks from stalling MQTT processing, and eliminates the handler-registry iteration race during reconnect/resubscribe - Iterate snapshots of handler registries so handlers can subscribe/unsubscribe during dispatch - Isolate individual handler failures: catch Exception per handler with logging instead of only (TypeError, AttributeError, KeyError) - Make the unit system preference process-wide instead of a ContextVar; tasks scheduled from CRT threads never inherited the caller's context, so --unit-system was silently ignored for all MQTT-delivered data - Remove once-listeners before invoking them: raising callbacks no longer stay registered forever and concurrent emits no longer double-fire - Clean up wait_for() listeners in a finally block so cancellation does not leak them; fix wait_for docstring examples that broke user code --- CHANGELOG.rst | 41 +++++ src/nwp500/events.py | 40 +++-- src/nwp500/mqtt/subscriptions.py | 64 ++++--- src/nwp500/mqtt_events.py | 2 +- src/nwp500/unit_system.py | 45 ++--- tests/test_threading_model.py | 299 +++++++++++++++++++++++++++++++ 6 files changed, 430 insertions(+), 61 deletions(-) create mode 100644 tests/test_threading_model.py diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 89741ba9..9eef031c 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -5,6 +5,47 @@ Changelog Unreleased ========== +Bug Fixes +--------- +- **Run MQTT message dispatch on the event loop**: JSON parsing, pydantic + model validation, and user callbacks all executed directly on the AWS + CRT network thread. A slow or blocking callback (e.g. the CLI monitor's + CSV writes) stalled all MQTT message processing, user callbacks ran on + an undocumented SDK thread where asyncio operations are unsafe, and the + handler registry could be mutated on the event loop while the CRT + thread iterated it (``RuntimeError: dictionary changed size during + iteration``, dropping the message — most likely during the + reconnect/resubscribe window). The awscrt callback now only marshals + the raw payload onto the event loop; parsing and dispatch run there, + iterating snapshots of the handler registries. +- **Stop one raising handler from aborting message delivery**: handler + dispatch caught only ``(TypeError, AttributeError, KeyError)``; a user + callback raising anything else (e.g. ``ValueError``) escaped into the + awscrt callback machinery and skipped the remaining handlers for that + message. Individual handler failures are now logged and isolated. +- **Make the unit system preference process-wide**: + ``set_unit_system()`` stored the preference in a ``ContextVar`` set in + the caller's task. Tasks scheduled from AWS CRT callback threads never + inherit that context, so the preference silently reverted to + auto-detect for all MQTT-delivered data — ``nwp-cli --unit-system + metric monitor`` logged temperatures in the device's native unit while + labeling them °C. The preference is now a process-wide setting visible + from every task and thread. +- **Fix once-listeners firing more than once**: ``EventEmitter.emit()`` + removed one-time listeners only after invoking them, so a callback + that raised stayed registered forever, and two overlapping emits could + both fire the same once-listener. Once-listeners are now removed + before invocation. +- **Fix wait_for() leaking its listener on cancellation**: + ``EventEmitter.wait_for()`` removed its listener only on timeout; a + cancelled waiter left the listener registered until the event next + fired, setting a result on a dead future. Cleanup now happens in a + ``finally`` block. +- **Fix wait_for() docstring examples**: examples showed + ``args, _ = await emitter.wait_for(...)``, but ``wait_for`` returns + just the args tuple — following the documented pattern raised + ``ValueError`` or silently mis-assigned. + Version 8.1.3 (2026-06-15) ========================== diff --git a/src/nwp500/events.py b/src/nwp500/events.py index b6dc46eb..c061f39f 100644 --- a/src/nwp500/events.py +++ b/src/nwp500/events.py @@ -245,7 +245,21 @@ async def emit(self, event: str, *args: Any, **kwargs: Any) -> int: listeners: list[EventListener] = self._listeners[event].copy() # Copy to allow modification during iteration called_count = 0 - listeners_to_remove: list[EventListener] = [] + + # Remove one-time listeners BEFORE invoking them (synchronously, + # in the same event-loop slice as the snapshot): + # - a raising callback must not stay registered forever + # - concurrent emits each snapshot the list; deferring removal + # until after (awaited) iteration double-fires once-listeners + for listener in listeners: + if listener.once: + if listener in self._listeners.get(event, []): + self._listeners[event].remove(listener) + self._once_callbacks.discard((event, listener.callback)) + + # Clean up if no listeners left + if event in self._listeners and not self._listeners[event]: + del self._listeners[event] for listener in listeners: try: @@ -257,11 +271,6 @@ async def emit(self, event: str, *args: Any, **kwargs: Any) -> int: called_count += 1 - # Check if this is a once listener - if listener.once: - listeners_to_remove.append(listener) - self._once_callbacks.discard((event, listener.callback)) - except Exception as e: # Catch all exceptions from user callbacks to ensure # resilience. We intentionally catch Exception here because: @@ -273,15 +282,6 @@ async def emit(self, event: str, *args: Any, **kwargs: Any) -> int: exc_info=True, ) - # Remove one-time listeners after iteration - for listener in listeners_to_remove: - if listener in self._listeners[event]: - self._listeners[event].remove(listener) - - # Clean up if no listeners left - if not self._listeners[event]: - del self._listeners[event] - # Track event count self._event_counts[event] += 1 @@ -391,7 +391,7 @@ async def wait_for( await emitter.wait_for('device_ready', timeout=30) # Wait for specific condition - args, _ = await emitter.wait_for('temperature_changed') + args = await emitter.wait_for('temperature_changed') temperature_event = args[0] current_temp = temperature_event.new_temperature """ @@ -416,7 +416,9 @@ def handler(*args: Any, **kwargs: Any) -> None: # Return just args for simplicity (most common case) return args_tuple - except TimeoutError: - # Remove the listener on timeout + finally: + # Always unregister: on timeout AND on cancellation. A leaked + # once-listener would otherwise stay registered until the + # event next fires (setting a result on a dead future). + # No-op when the listener already fired (once() removed it). self.off(event, handler) - raise diff --git a/src/nwp500/mqtt/subscriptions.py b/src/nwp500/mqtt/subscriptions.py index 52cf1d82..37a243bf 100644 --- a/src/nwp500/mqtt/subscriptions.py +++ b/src/nwp500/mqtt/subscriptions.py @@ -118,37 +118,61 @@ def update_connection(self, connection: Any) -> None: def _on_message_received( self, topic: str, payload: bytes, **kwargs: Any ) -> None: - """Handle received MQTT messages. + """Handle received MQTT messages (AWS CRT callback). - Parses JSON payload and routes to registered handlers. + This runs on the AWS CRT network thread. It must stay trivial: + JSON parsing, pydantic validation, and user callbacks are + marshaled onto the asyncio event loop so that + + - a slow or blocking user callback cannot stall all MQTT message + processing, + - handler registries mutated on the event loop are never iterated + concurrently from this thread, and + - user callbacks run where asyncio operations are safe, instead + of on an undocumented SDK thread. Args: topic: MQTT topic the message was received on payload: Raw message payload (JSON bytes) **kwargs: Additional MQTT metadata """ + self._schedule_coroutine(self._dispatch_message(topic, payload)) + + async def _dispatch_message(self, topic: str, payload: bytes) -> None: + """Parse a message and route it to registered handlers. + + Runs on the event loop. + + Args: + topic: MQTT topic the message was received on + payload: Raw message payload (JSON bytes) + """ try: - # Parse JSON payload message = json.loads(payload.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + _logger.error(f"Failed to parse message payload: {e}") + return + + if _logger.isEnabledFor(logging.DEBUG): _logger.debug("Received message on topic: %s", redact_topic(topic)) - # Call registered handlers that match this topic - # Need to match against subscription patterns with wildcards - for ( - subscription_pattern, - handlers, - ) in self._message_handlers.items(): - if topic_matches_pattern(topic, subscription_pattern): - for handler in handlers: - try: - handler(topic, message) - except (TypeError, AttributeError, KeyError) as e: - _logger.error(f"Error in message handler: {e}") - - except json.JSONDecodeError as e: - _logger.error(f"Failed to parse message payload: {e}") - except (AttributeError, KeyError, TypeError) as e: - _logger.error(f"Error processing message: {e}") + # Call registered handlers that match this topic. + # Iterate over snapshots so handlers can subscribe/unsubscribe + # (mutating the registries) during dispatch. + for subscription_pattern, handlers in list( + self._message_handlers.items() + ): + if topic_matches_pattern(topic, subscription_pattern): + for handler in list(handlers): + try: + handler(topic, message) + except Exception: + # One bad handler must not abort delivery to the + # remaining handlers for this message. + _logger.exception( + "Error in message handler for %s", + redact_topic(topic), + ) async def subscribe( self, diff --git a/src/nwp500/mqtt_events.py b/src/nwp500/mqtt_events.py index 22aeac63..bdd6257c 100644 --- a/src/nwp500/mqtt_events.py +++ b/src/nwp500/mqtt_events.py @@ -204,7 +204,7 @@ class MqttClientEvents: ) # Wait for a specific event - args, _ = await mqtt_client.wait_for( + args = await mqtt_client.wait_for( MqttClientEvents.CONNECTION_RESUMED ) connection_event = args[0] diff --git a/src/nwp500/unit_system.py b/src/nwp500/unit_system.py index 3babef8b..721311cf 100644 --- a/src/nwp500/unit_system.py +++ b/src/nwp500/unit_system.py @@ -1,16 +1,18 @@ """Unit system management for temperature, flow rate, and volume conversions. -This module provides context-based unit system management, allowing applications -to override the device's temperature_type setting and specify a preferred -measurement system (Metric or Imperial). +This module provides process-wide unit system management, allowing +applications to override the device's temperature_type setting and specify a +preferred measurement system (Metric or Imperial). The unit system preference can be set at library initialization and is used -during model validation to convert device values to the user's preferred units. +during model validation to convert device values to the user's preferred +units. The preference is a process-wide setting: it is visible from every +task and thread, including model validation triggered by MQTT callbacks that +run outside the task that configured it. """ from __future__ import annotations -import contextvars import logging from typing import Literal @@ -25,12 +27,15 @@ # Type alias for unit system preference UnitSystemType = Literal["metric", "us_customary"] | None -# Context variable to store the preferred unit system +# Process-wide preferred unit system. # None means auto-detect from device # "metric" means Celsius, "us_customary" means Fahrenheit -_unit_system_context: contextvars.ContextVar[ - Literal["metric", "us_customary"] | None -] = contextvars.ContextVar("unit_system", default=None) +# +# This is intentionally NOT a contextvars.ContextVar: MQTT message handling +# runs in tasks scheduled from AWS CRT callback threads whose context never +# inherits from the application task, so a context-local preference silently +# reverted to auto-detect for all real-time data. +_unit_system: Literal["metric", "us_customary"] | None = None def set_unit_system( @@ -38,8 +43,9 @@ def set_unit_system( ) -> None: """Set preferred unit system for temperature, flow, and volume conversions. - This setting overrides the device's temperature_type setting and applies to - all subsequent model validation operations in the current async context. + This setting overrides the device's temperature_type setting and applies + process-wide to all subsequent model validation operations, including + those triggered by MQTT callbacks running in other tasks or threads. Args: unit_system: Preferred unit system: @@ -52,12 +58,9 @@ def set_unit_system( >>> set_unit_system("us_customary") >>> # All values now in F, GPM, Gallons >>> set_unit_system(None) # Reset to auto-detect - - Note: - This is context-aware and works with async code. Each async task - maintains its own unit system preference. """ - _unit_system_context.set(unit_system) + global _unit_system + _unit_system = unit_system def get_unit_system() -> Literal["metric", "us_customary"] | None: @@ -69,7 +72,7 @@ def get_unit_system() -> Literal["metric", "us_customary"] | None: - "us_customary": Fahrenheit, GPM, Gallons - None: Auto-detect from device (default) """ - return _unit_system_context.get() + return _unit_system def reset_unit_system() -> None: @@ -78,7 +81,7 @@ def reset_unit_system() -> None: This is useful for tests or when switching between different device configurations. """ - _unit_system_context.set(None) + set_unit_system(None) def unit_system_to_temperature_type( @@ -108,8 +111,8 @@ def is_metric_preferred( ) -> bool: """Check if metric (Celsius) is preferred. - Checks the override first, then falls back to the context-configured - unit system. Used during validation to determine preferred units. + Checks the override first, then falls back to the configured unit + system. Used during validation to determine preferred units. Args: override: Optional override value. If provided, this takes precedence @@ -123,7 +126,7 @@ def is_metric_preferred( if override is not None: return override == "metric" - # Otherwise check context + # Otherwise check the configured preference unit_system = get_unit_system() if unit_system is not None: return unit_system == "metric" diff --git a/tests/test_threading_model.py b/tests/test_threading_model.py new file mode 100644 index 00000000..fd922601 --- /dev/null +++ b/tests/test_threading_model.py @@ -0,0 +1,299 @@ +"""Regression tests for MQTT threading model and event emitter fixes. + +Covers: +- MQTT message dispatch marshaled from the AWS CRT thread to the event loop +- Handler registry mutation during dispatch (snapshot iteration) +- One raising handler not aborting delivery to remaining handlers +- Unit system preference visible across tasks and threads +- Once-listeners removed even when their callback raises +- Concurrent emits not double-firing once-listeners +- wait_for() cleaning up its listener on cancellation +""" + +from __future__ import annotations + +import asyncio +import json +import threading + +import pytest + +from nwp500.events import EventEmitter +from nwp500.mqtt.subscriptions import MqttSubscriptionManager +from nwp500.unit_system import ( + get_unit_system, + reset_unit_system, + set_unit_system, +) + + +@pytest.fixture(autouse=True) +def _reset_units(): + yield + reset_unit_system() + + +def _make_manager(loop: asyncio.AbstractEventLoop) -> MqttSubscriptionManager: + def schedule(coro): + asyncio.run_coroutine_threadsafe(coro, loop) + + return MqttSubscriptionManager( + connection=object(), + client_id="test-client", + event_emitter=EventEmitter(), + schedule_coroutine=schedule, + ) + + +class TestDispatchThreading: + """Message parse/dispatch must run on the event loop, not the CRT + thread.""" + + @pytest.mark.asyncio + async def test_message_from_foreign_thread_dispatches_on_loop(self): + """Regression: JSON decode, model validation, and user callbacks + all ran on the AWS CRT network thread — blocking callbacks + stalled MQTT processing and shared state was touched off-loop.""" + loop = asyncio.get_running_loop() + manager = _make_manager(loop) + + received = asyncio.Event() + seen: dict = {} + + def handler(topic, message): + seen["thread"] = threading.get_ident() + seen["message"] = message + loop.call_soon(received.set) + + manager._message_handlers["cmd/52/+/st"] = [handler] + + payload = json.dumps({"key": "value"}).encode() + + # Simulate the awscrt callback arriving on a foreign thread + crt_thread = threading.Thread( + target=manager._on_message_received, + args=("cmd/52/navilink-test/st", payload), + ) + crt_thread.start() + crt_thread.join() + + await asyncio.wait_for(received.wait(), timeout=2.0) + + assert seen["message"] == {"key": "value"} + assert seen["thread"] == threading.get_ident() # event loop thread + + @pytest.mark.asyncio + async def test_handler_mutating_registry_during_dispatch(self): + """Regression: dispatch iterated the live handler dict; a + subscribe/unsubscribe during delivery raised 'dictionary changed + size during iteration' and dropped the message.""" + loop = asyncio.get_running_loop() + manager = _make_manager(loop) + + calls = [] + + def mutating_handler(topic, message): + calls.append("mutating") + # Mutate both registries mid-dispatch + manager._message_handlers["cmd/52/other/topic"] = [ + lambda t, m: None + ] + manager._message_handlers.pop("cmd/52/+/st", None) + + def second_handler(topic, message): + calls.append("second") + + manager._message_handlers["cmd/52/+/st"] = [ + mutating_handler, + second_handler, + ] + + await manager._dispatch_message( + "cmd/52/navilink-test/st", json.dumps({}).encode() + ) + + assert calls == ["mutating", "second"] + + @pytest.mark.asyncio + async def test_raising_handler_does_not_block_others(self): + """Regression: only (TypeError, AttributeError, KeyError) were + caught; a handler raising ValueError escaped into the awscrt + callback machinery and aborted delivery to remaining handlers.""" + loop = asyncio.get_running_loop() + manager = _make_manager(loop) + + calls = [] + + def bad_handler(topic, message): + raise ValueError("boom") + + def good_handler(topic, message): + calls.append("good") + + manager._message_handlers["cmd/52/+/st"] = [bad_handler, good_handler] + + await manager._dispatch_message( + "cmd/52/navilink-test/st", json.dumps({}).encode() + ) + + assert calls == ["good"] + + @pytest.mark.asyncio + async def test_invalid_json_is_logged_not_raised(self, caplog): + loop = asyncio.get_running_loop() + manager = _make_manager(loop) + manager._message_handlers["t"] = [lambda t, m: None] + + # Must not raise + await manager._dispatch_message("t", b"{not json") + + assert "Failed to parse message payload" in caplog.text + + +class TestUnitSystemVisibility: + """The unit system preference must be process-wide.""" + + @pytest.mark.asyncio + async def test_visible_in_task_scheduled_from_foreign_thread(self): + """Regression: the preference was a ContextVar set in the main + task; tasks scheduled via run_coroutine_threadsafe from the CRT + thread never inherited it, so --unit-system was silently ignored + for all MQTT-delivered data.""" + loop = asyncio.get_running_loop() + set_unit_system("metric") + + result: dict = {} + done = asyncio.Event() + + async def read_preference(): + result["unit_system"] = get_unit_system() + done.set() + + def foreign_thread(): + asyncio.run_coroutine_threadsafe(read_preference(), loop) + + thread = threading.Thread(target=foreign_thread) + thread.start() + thread.join() + + await asyncio.wait_for(done.wait(), timeout=2.0) + assert result["unit_system"] == "metric" + + def test_visible_across_plain_threads(self): + set_unit_system("us_customary") + seen = {} + + def reader(): + seen["value"] = get_unit_system() + + thread = threading.Thread(target=reader) + thread.start() + thread.join() + + assert seen["value"] == "us_customary" + + def test_reset_restores_auto_detect(self): + set_unit_system("metric") + reset_unit_system() + assert get_unit_system() is None + + +class TestOnceListeners: + """Once-listeners must fire exactly once, no matter what.""" + + @pytest.mark.asyncio + async def test_raising_once_listener_is_still_removed(self): + """Regression: removal happened after the callback inside the + try block; a raising callback skipped removal and fired on every + subsequent emit.""" + emitter = EventEmitter() + calls = [] + + def bad_handler(): + calls.append(1) + raise ValueError("boom") + + emitter.once("evt", bad_handler) + + await emitter.emit("evt") + await emitter.emit("evt") + + assert len(calls) == 1 + assert emitter.listener_count("evt") == 0 + + @pytest.mark.asyncio + async def test_concurrent_emits_fire_once_listener_once(self): + """Regression: removal was deferred until after the (awaited) + iteration; overlapping emits each snapshotted the listener list + before either removed, double-firing once-listeners.""" + emitter = EventEmitter() + calls = [] + + async def slow_handler(): + calls.append(1) + await asyncio.sleep(0.01) + + emitter.once("evt", slow_handler) + + await asyncio.gather(emitter.emit("evt"), emitter.emit("evt")) + + assert len(calls) == 1 + + @pytest.mark.asyncio + async def test_regular_listeners_survive_emit(self): + emitter = EventEmitter() + calls = [] + emitter.on("evt", lambda: calls.append(1)) + + await emitter.emit("evt") + await emitter.emit("evt") + + assert len(calls) == 2 + assert emitter.listener_count("evt") == 1 + + +class TestWaitForCleanup: + """wait_for() must never leak its listener.""" + + @pytest.mark.asyncio + async def test_cancellation_removes_listener(self): + """Regression: only TimeoutError removed the listener; a + cancelled waiter left it registered until the event next fired, + setting a result on a dead future.""" + emitter = EventEmitter() + + task = asyncio.create_task(emitter.wait_for("evt")) + await asyncio.sleep(0.01) # let the listener register + assert emitter.listener_count("evt") == 1 + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert emitter.listener_count("evt") == 0 + # A later emit must not blow up on the dead future + await emitter.emit("evt") + + @pytest.mark.asyncio + async def test_timeout_removes_listener(self): + emitter = EventEmitter() + + with pytest.raises(TimeoutError): + await emitter.wait_for("evt", timeout=0.01) + + assert emitter.listener_count("evt") == 0 + + @pytest.mark.asyncio + async def test_successful_wait_returns_args(self): + emitter = EventEmitter() + + async def fire(): + await asyncio.sleep(0.01) + await emitter.emit("evt", "payload") + + fire_task = asyncio.create_task(fire()) + args = await emitter.wait_for("evt", timeout=1.0) + await fire_task + + assert args == ("payload",) + assert emitter.listener_count("evt") == 0 From 42c2d2299a82f3bc7450653f1299651043ce5874 Mon Sep 17 00:00:00 2001 From: emmanuel Date: Sun, 5 Jul 2026 09:36:39 -0700 Subject: [PATCH 2/2] test: reset unit system before and after each threading-model test Addresses review feedback: the preference is process-wide, so a teardown-only reset let the first test in this module inherit values set by earlier tests in the session. --- tests/test_threading_model.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/test_threading_model.py b/tests/test_threading_model.py index fd922601..f183d601 100644 --- a/tests/test_threading_model.py +++ b/tests/test_threading_model.py @@ -29,6 +29,12 @@ @pytest.fixture(autouse=True) def _reset_units(): + """Reset the process-wide unit preference before AND after each test. + + Resetting only in teardown would let the first test in this module + inherit a value set by earlier tests in the session. + """ + reset_unit_system() yield reset_unit_system()