Add exit on exception for deamon threads

Author: avillega

resonate/bridge.py

@@ -35,6 +35,7 @@
 from resonate.models.task import Task
 from resonate.options import Options
 from resonate.scheduler import Done, Info, More, Scheduler
+from resonate.utils import exit_on_exception
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -176,6 +177,7 @@ def stop(self) -> None:
         if self._heartbeat_thread.is_alive():
             self._heartbeat_thread.join()
 
+    @exit_on_exception
     def _process_cq(self) -> None:
         while True:
             item = self._cq.get()
@@ -242,6 +244,7 @@ def _process_cq(self) -> None:
                             if task is not None:
                                 self._store.tasks.complete(id=task.id, counter=task.counter)
 
+    @exit_on_exception
     def _process_msgs(self) -> None:
         def _invoke(root: DurablePromise) -> Invoke:
             assert "func" in root.param.data
@@ -301,6 +304,7 @@ def _invoke(root: DurablePromise) -> Invoke:
                     durable_promise = DurablePromise.from_dict(self._store, _promise)
                     self._cq.put_nowait(Notify(durable_promise.id, durable_promise))
 
+    @exit_on_exception
     def _process_delayed_events(self) -> None:
         while not self._shutdown.is_set():
             with self._delay_condition:
@@ -337,6 +341,7 @@ def _process_delayed_events(self) -> None:
     def start_heartbeat(self) -> None:
         self._heartbeat_active.set()
 
+    @exit_on_exception
     def _heartbeat(self) -> None:
         while not self._shutdown.is_set():
             # If this timeout don't execute the heartbeat

resonate/message_sources/local.py

@@ -3,6 +3,8 @@
 import threading
 from typing import TYPE_CHECKING
 
+from resonate.utils import exit_on_exception
+
 if TYPE_CHECKING:
     from resonate.models.message import Mesg
     from resonate.models.message_source import MessageQ
@@ -45,6 +47,7 @@ def stop(self) -> None:
             self._thread = None
             self._stop_event.clear()
 
+    @exit_on_exception
     def _loop(self, mq: MessageQ) -> None:
         while not self._stop_event.is_set():
             for msg in self.step():

resonate/message_sources/poller.py

@@ -10,6 +10,7 @@
 from resonate.encoders import JsonEncoder
 from resonate.logging import logger
 from resonate.models.message import InvokeMesg, NotifyMesg, ResumeMesg
+from resonate.utils import exit_on_exception
 
 if TYPE_CHECKING:
     from collections.abc import Generator
@@ -69,6 +70,7 @@ def stop(self) -> None:
         # value, which is not the default. This is still useful for tests.
         self._shutdown = True
 
+    @exit_on_exception
     def loop(self, mq: MessageQ) -> None:
         while True:
             if self._shutdown:
@@ -91,7 +93,7 @@ def loop(self, mq: MessageQ) -> None:
                 continue
             except Exception as e:
                 logger.warning("Unexpected error in poller loop for group %s: %s", self._group, e)
-                break
+                raise
 
     def step(self) -> list[Mesg]:
         with requests.get(self.url, stream=True, timeout=self._timeout) as res:

resonate/utils.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import os
+import traceback
+import urllib.parse
+from functools import wraps
+from importlib.metadata import version
+from typing import TYPE_CHECKING, ParamSpec, TypeVar
+
+from resonate.logging import logger
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+def exit_on_exception(func: Callable[P, R]) -> Callable[P, R]:
+    @wraps(func)
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        try:
+            return func(*args, **kwargs)
+        except Exception:
+            v = version("resonate-sdk")
+            logger.exception(
+                "An unexpected error happened.\n\nPlease report this issue so we can fix it as fast a possible:\n - https://github.com/resonatehq/resonate-sdk-py/issues/new?body=%s\n\n",
+                urllib.parse.quote(f"Resonate (version {v}) process exited with error:\n```bash\n{traceback.format_exc()}```"),
+            )
+            os._exit(1)
+
+    return wrapper