Repository: XKNX/xknx
Pull request #767 (2021-10-01 20:14:16)

Catch and log exceptions from processed Telegrams
<!--
  You are awesome! Thanks for contributing to our project!
  Please, DO NOT DELETE ANY TEXT from this template!.
-->
## Description
<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.
-->
When handling a Telegram (received or sent) raises an Exception - from a Callback or an Error in xknx -  it should not cancel the TelegramQueue.
These unexpected errors will now be logged and the TelegramQueue keeps processing telegrams.

Callbacks for devices and telegrams are now handled in separate asyncio Tasks (through `asyncio.gather()`) which may enable some async concurrency but also adds a little overhead 🙃 (in my local tests - with `timeit` not real-world - it was a little bit slower than the previous loop-await implementation - but nothing significant).


## Type of change
<!--
Please tick the applicable options.
NOTE: Ticking multiple options most likely indicates
that your change is to big and it is suggested to split it into several smaller PRs.
-->

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## Checklist:

- [x] I have performed a self-review of my own code
- [ ] The documentation has been adjusted accordingly
- [ ] The changes generate no new warnings
- [x] Tests have been added that prove the fix is effective or that the feature works
- [x] The changes are documented in the changelog
- [ ] The Homeassistant plugin has been adjusted in case of new config options


codecov[bot]: # [Codecov](https://codecov.io/gh/XKNX/xknx/pull/767?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) Report
> Merging [#767](https://codecov.io/gh/XKNX/xknx/pull/767?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (ddd322f) into [main](https://codecov.io/gh/XKNX/xknx/commit/c19ea3d4347513d6052f358ea98948e248a68e97?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) (c19ea3d) will **decrease** coverage by `0.05%`.
> The diff coverage is `76.47%`.

[![Impacted file tree graph](https://codecov.io/gh/XKNX/xknx/pull/767/graphs/tree.svg?width=650&height=150&src=pr&token=irWbIygS84&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)](https://codecov.io/gh/XKNX/xknx/pull/767?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX)

```diff
@@            Coverage Diff             @@
##             main     #767      +/-   ##
==========================================
- Coverage   94.67%   94.61%   -0.06%     
==========================================
  Files         106      106              
  Lines        6663     6672       +9     
==========================================
+ Hits         6308     6313       +5     
- Misses        355      359       +4     
```


| [Impacted Files](https://codecov.io/gh/XKNX/xknx/pull/767?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX) | Coverage Δ | |
|---|---|---|
| [xknx/core/telegram\_queue.py](https://codecov.io/gh/XKNX/xknx/pull/767/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9jb3JlL3RlbGVncmFtX3F1ZXVlLnB5) | `92.10% <66.66%> (-3.27%)` | :arrow_down: |
| [xknx/devices/device.py](https://codecov.io/gh/XKNX/xknx/pull/767/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=XKNX#diff-eGtueC9kZXZpY2VzL2RldmljZS5weQ==) | `100.00% <100.00%> (ø)` | |


Patch:
diff --git a/changelog.md b/changelog.md
index 6011bc92..412b0484 100644
--- a/changelog.md
+++ b/changelog.md
@@ -4,6 +4,8 @@
 
 ### Internals
 
+- Catch and log exceptions raised in callbacks to not stall the TelegramQueue
+- Handle callbacks in separate asyncio Tasks
 - GatewayScanFilter: Ignore non-gateway KNX/IP devices
 
 ## 0.18.9 HS-color 2021-07-26
diff --git a/xknx/core/telegram_queue.py b/xknx/core/telegram_queue.py
index bd4a4e2c..1e019149 100644
--- a/xknx/core/telegram_queue.py
+++ b/xknx/core/telegram_queue.py
@@ -126,6 +126,12 @@ class TelegramQueue:
                     await self.process_telegram_incoming(telegram)
                 except XKNXException as ex:
                     logger.error("Error while processing incoming telegram %s", ex)
+                except Exception:  # pylint: disable=broad-except
+                    # prevent the parser Task from stalling when unexpected errors occur
+                    logger.exception(
+                        "Unexpected error while processing incoming telegram %s",
+                        telegram,
+                    )
                 finally:
                     self.xknx.telegrams.task_done()
             elif telegram.direction == TelegramDirection.OUTGOING:
@@ -148,6 +154,11 @@ class TelegramQueue:
                     logger.warning(ex)
             except XKNXException as ex:
                 logger.error("Error while processing outgoing telegram %s", ex)
+            except Exception:  # pylint: disable=broad-except
+                # prevent the sender Task from stalling when unexpected errors occur (eg. ValueError from creating KNXIPFrames)
+                logger.exception(
+                    "Unexpected error while processing outgoing telegram %s", telegram
+                )
             finally:
                 self.outgoing_queue.task_done()
                 self.xknx.telegrams.task_done()
@@ -183,14 +194,25 @@ class TelegramQueue:
             await self.xknx.knxip_interface.send_telegram(telegram)
 
         await self.xknx.devices.process(telegram)
-        for telegram_received_cb in self.telegram_received_cbs:
-            if telegram_received_cb.is_within_filter(telegram):
-                await telegram_received_cb.callback(telegram)
+        await self._run_telegram_received_cbs(telegram)
 
     async def process_telegram_incoming(self, telegram: Telegram) -> None:
         """Process incoming telegram."""
         telegram_logger.debug(telegram)
-        for telegram_received_cb in self.telegram_received_cbs:
-            if telegram_received_cb.is_within_filter(telegram):
-                await telegram_received_cb.callback(telegram)
+        await self._run_telegram_received_cbs(telegram)
         await self.xknx.devices.process(telegram)
+
+    async def _run_telegram_received_cbs(self, telegram: Telegram) -> None:
+        """Run registered callbacks. Don't propagate exceptions."""
+        callbacks = [
+            cb.callback(telegram)
+            for cb in self.telegram_received_cbs
+            if cb.is_within_filter(telegram)
+        ]
+        try:
+            await asyncio.gather(*callbacks)
+        except Exception:  # pylint: disable=broad-except
+            logger.exception(
+                "Unexpected error while processing telegram_received_cb for %s",
+                telegram,
+            )
diff --git a/xknx/devices/device.py b/xknx/devices/device.py
index d9e42155..b862e481 100644
--- a/xknx/devices/device.py
+++ b/xknx/devices/device.py
@@ -6,6 +6,7 @@ It provides basis functionality for reading the state from the KNX bus.
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+import asyncio
 import logging
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterator
 
@@ -73,8 +74,13 @@ class Device(ABC):
 
     async def after_update(self) -> None:
         """Execute callbacks after internal state has been changed."""
-        for device_updated_cb in self.device_updated_cbs:
-            await device_updated_cb(self)
+        try:
+            await asyncio.gather(*[cb(self) for cb in self.device_updated_cbs])
+        except Exception:  # pylint: disable=broad-except
+            logger.exception(
+                "Unexpected error while processing device_updated_cb for %s",
+                self,
+            )
 
     async def sync(self, wait_for_result: bool = False) -> None:
         """Read states of device from KNX bus."""


Test patch:
diff --git a/test/core_tests/telegram_queue_test.py b/test/core_tests/telegram_queue_test.py
index 74481726..10016bb4 100644
--- a/test/core_tests/telegram_queue_test.py
+++ b/test/core_tests/telegram_queue_test.py
@@ -1,6 +1,6 @@
 """Unit test for telegram received callback."""
 import asyncio
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
 from xknx import XKNX
@@ -238,7 +238,7 @@ class TestTelegramQueue:
 
         async def process_exception():
             raise CouldNotParseTelegram(
-                "Something went wrong when receiving the telegram." ""
+                "Something went wrong when receiving the telegram."
             )
 
         process_tg_in_mock.return_value = asyncio.ensure_future(process_exception())
@@ -254,9 +254,7 @@ class TestTelegramQueue:
 
         logging_error_mock.assert_called_once_with(
             "Error while processing telegram %s",
-            CouldNotParseTelegram(
-                "Something went wrong when receiving the telegram." ""
-            ),
+            CouldNotParseTelegram("Something went wrong when receiving the telegram."),
         )
 
     @patch("xknx.core.TelegramQueue.process_telegram_outgoing", new_callable=AsyncMock)
@@ -385,3 +383,68 @@ class TestTelegramQueue:
         await xknx.telegram_queue.process_telegram_incoming(telegram)
         async_telegram_received_cb_one.assert_not_called()
         async_telegram_received_cb_two.assert_called_once_with(telegram)
+
+    #
+    # TEST EXCEPTION HANDLING
+    #
+    @patch("logging.Logger.exception")
+    @patch("xknx.xknx.Devices.process", side_effect=Exception)
+    async def test_process_raising(self, process_mock, logging_exception_mock):
+        """Test unexpected exception handling in telegram queues."""
+        xknx = XKNX()
+        telegram_in = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            direction=TelegramDirection.INCOMING,
+            payload=GroupValueWrite(DPTBinary(1)),
+        )
+        # InternalGroupAddress to avoid CommunicationError for missing knxip_interface
+        telegram_out = Telegram(
+            destination_address=InternalGroupAddress("i-outgoing"),
+            direction=TelegramDirection.OUTGOING,
+            payload=GroupValueWrite(DPTBinary(0)),
+        )
+        xknx.telegrams.put_nowait(telegram_in)
+        xknx.telegrams.put_nowait(telegram_out)
+
+        await xknx.telegram_queue.start()
+        await xknx.telegram_queue.stop()
+
+        log_calls = [
+            call(
+                "Unexpected error while processing incoming telegram %s",
+                telegram_in,
+            ),
+            call(
+                "Unexpected error while processing outgoing telegram %s",
+                telegram_out,
+            ),
+        ]
+        logging_exception_mock.assert_has_calls(log_calls)
+
+    @patch("logging.Logger.exception")
+    async def test_callback_raising(self, logging_exception_mock):
+        """Test telegram_received_callback raising an exception."""
+        xknx = XKNX()
+        good_callback_1 = AsyncMock()
+        bad_callback = AsyncMock(side_effect=Exception("Boom"))
+        good_callback_2 = AsyncMock()
+
+        xknx.telegram_queue.register_telegram_received_cb(good_callback_1)
+        xknx.telegram_queue.register_telegram_received_cb(bad_callback)
+        xknx.telegram_queue.register_telegram_received_cb(good_callback_2)
+
+        telegram = Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            direction=TelegramDirection.INCOMING,
+            payload=GroupValueWrite(DPTBinary(1)),
+        )
+        await xknx.telegram_queue.process_telegram_incoming(telegram)
+
+        good_callback_1.assert_called_once_with(telegram)
+        bad_callback.assert_called_once_with(telegram)
+        good_callback_2.assert_called_once_with(telegram)
+
+        logging_exception_mock.assert_called_once_with(
+            "Unexpected error while processing telegram_received_cb for %s",
+            telegram,
+        )
diff --git a/test/devices_tests/device_test.py b/test/devices_tests/device_test.py
index eb20f4e7..d853fb8a 100644
--- a/test/devices_tests/device_test.py
+++ b/test/devices_tests/device_test.py
@@ -60,6 +60,29 @@ class TestDevice:
         after_update_callback1.assert_not_called()
         after_update_callback2.assert_not_called()
 
+    @patch("logging.Logger.exception")
+    async def test_bad_callback(self, logging_exception_mock):
+        """Test handling callback raising an exception"""
+        xknx = XKNX()
+        device = Device(xknx, "TestDevice")
+        good_callback_1 = AsyncMock()
+        bad_callback = AsyncMock(side_effect=Exception("Boom"))
+        good_callback_2 = AsyncMock()
+
+        device.register_device_updated_cb(good_callback_1)
+        device.register_device_updated_cb(bad_callback)
+        device.register_device_updated_cb(good_callback_2)
+
+        await device.after_update()
+        good_callback_1.assert_called_with(device)
+        bad_callback.assert_called_with(device)
+        good_callback_2.assert_called_with(device)
+
+        logging_exception_mock.assert_called_once_with(
+            "Unexpected error while processing device_updated_cb for %s",
+            device,
+        )
+
     async def test_process(self):
         """Test if telegram is handled by the correct process_* method."""
         xknx = XKNX()