Fix crash and silent-failure bugs in TDJson, listener, and worker

- TDJson.__del__ checked hasattr on the wrong object (CDLL instead of
  self), so the destructor never called stop(). Fix the hasattr target.
- ctypes fatal error callback was a local variable eligible for GC while
  TDLib still held its pointer. Store it on self to prevent collection.
- TDJson.stop() had no guard against double-destroy: null out the client
  handle after destroying and early-return if already None.
- Worker thread had no try/except around handler calls — a single
  failing handler killed the daemon thread silently.
- Listener thread had no exception handling at all — any error in
  receive/update/handlers killed it, hanging all pending AsyncResults.
- queue.Full from _run_handlers propagated into the listener, creating a
  cascading failure when the worker died.
- send_message with Element used assert for error checking, which is
  stripped under python -O. Use raise_exc=True and explicit RuntimeError.

Author: alexander-akhmetov

telegram/client.py

@@ -254,8 +254,9 @@ def send_message(
         updated_text: str
         if isinstance(text, Element):
             result = self.parse_text_entities(text.to_html(), parse_mode="HTML")
-            result.wait()
-            assert result.update is not None
+            result.wait(raise_exc=True)
+            if result.update is None:
+                raise RuntimeError(f"Failed to parse text entities: {result.error_info}")
             update: dict = result.update
             entities = update["entities"]
             updated_text = update["text"]
@@ -524,11 +525,16 @@ def _listen_to_td(self) -> None:
         logger.info("[Telegram.td_listener] started")
 
         while not self._stopped.is_set():
-            update = self._tdjson.receive()
+            try:
+                update = self._tdjson.receive()
 
-            if update:
-                self._update_async_result(update)
-                self._run_handlers(update)
+                if update:
+                    self._update_async_result(update)
+                    self._run_handlers(update)
+            except Exception:
+                if self._stopped.is_set():
+                    break
+                logger.exception("[Telegram.td_listener] error processing update")
 
     def _update_async_result(self, update: Dict[Any, Any]) -> typing.Optional[AsyncResult]:
         async_result = None
@@ -559,7 +565,10 @@ def _run_handlers(self, update: Dict[Any, Any]) -> None:
         update_type: str = update.get("@type", "unknown")
 
         for handler in self._update_handlers[update_type]:
-            self._workers_queue.put((handler, update), timeout=self._queue_put_timeout)
+            try:
+                self._workers_queue.put((handler, update), timeout=self._queue_put_timeout)
+            except queue.Full:
+                logger.error("Handler queue full, dropping update %s for handler %s", update_type, handler)
 
     def remove_update_handler(self, handler_type: str, func: Callable) -> None:
         """

telegram/tdjson.py

@@ -32,7 +32,7 @@ def __init__(self, library_path: Optional[str] = None, verbosity: int = 2) -> No
         self._build_client(library_path, verbosity)
 
     def __del__(self) -> None:
-        if hasattr(self, "_tdjson") and hasattr(self._tdjson, "_td_json_client_destroy"):
+        if hasattr(self, "_td_json_client_destroy"):
             self.stop()
 
     def _build_client(self, library_path: str, verbosity: int) -> None:
@@ -85,8 +85,8 @@ def _build_client(self, library_path: str, verbosity: int) -> None:
         def on_fatal_error_callback(error_message: str) -> None:
             logger.error("TDLib fatal error: %s", error_message)
 
-        c_on_fatal_error_callback = fatal_error_callback_type(on_fatal_error_callback)
-        self._td_set_log_fatal_error_callback(c_on_fatal_error_callback)
+        self._c_on_fatal_error_callback = fatal_error_callback_type(on_fatal_error_callback)
+        self._td_set_log_fatal_error_callback(self._c_on_fatal_error_callback)
 
     def send(self, query: Dict[Any, Any]) -> None:
         dumped_query = json.dumps(query).encode("utf-8")
@@ -116,4 +116,7 @@ def td_execute(self, query: Dict[Any, Any]) -> Union[Dict[Any, Any], Any]:
         return None
 
     def stop(self) -> None:
+        if self.td_json_client is None:
+            return
         self._td_json_client_destroy(self.td_json_client)
+        self.td_json_client = None

telegram/worker.py

@@ -41,7 +41,10 @@ def _run_thread(self) -> None:
             except Empty:
                 continue
 
-            handler(update)
+            try:
+                handler(update)
+            except Exception:
+                logger.exception("Error in update handler %s", handler)
             self._queue.task_done()
 
     def stop(self) -> None:

tests/test_tdjson.py

@@ -1,6 +1,6 @@
 from unittest.mock import Mock, patch
 
-from telegram.tdjson import _get_tdjson_lib_path
+from telegram.tdjson import TDJson, _get_tdjson_lib_path
 
 
 class TestGetTdjsonTdlibPath:
@@ -42,3 +42,40 @@ def test_unknown(self):
 
         mocked_files.assert_called_once_with("telegram")
         mocked_joinpath.assert_called_once_with("lib/linux/libtdjson.so")
+
+
+class TestTDJson:
+    def _make_tdjson(self):
+        with patch("telegram.tdjson.CDLL") as mocked_cdll:
+            mocked_cdll.return_value.td_json_client_create.return_value = 12345
+            tdjson = TDJson(library_path="/fake/lib.so", verbosity=0)
+        return tdjson
+
+    def test_del_calls_stop(self):
+        tdjson = self._make_tdjson()
+        with patch.object(tdjson, "stop") as mocked_stop:
+            tdjson.__del__()
+        mocked_stop.assert_called_once()
+
+    def test_del_skips_stop_if_build_incomplete(self):
+        tdjson = TDJson.__new__(TDJson)
+        with patch.object(TDJson, "stop") as mocked_stop:
+            tdjson.__del__()
+        mocked_stop.assert_not_called()
+
+    def test_stop_nulls_client_handle(self):
+        tdjson = self._make_tdjson()
+        assert tdjson.td_json_client is not None
+        tdjson.stop()
+        assert tdjson.td_json_client is None
+
+    def test_stop_is_idempotent(self):
+        tdjson = self._make_tdjson()
+        tdjson.stop()
+        tdjson.stop()
+        tdjson._td_json_client_destroy.assert_called_once()
+
+    def test_fatal_error_callback_stored_on_instance(self):
+        tdjson = self._make_tdjson()
+        assert hasattr(tdjson, "_c_on_fatal_error_callback")
+        assert tdjson._c_on_fatal_error_callback is not None

tests/test_telegram_methods.py

@@ -1,3 +1,5 @@
+import queue
+
 import pytest
 
 from unittest.mock import patch
@@ -478,3 +480,97 @@ def _get_async_result(data, request_id=None):
         assert state == telegram.authorization_state == AuthorizationState.READY
 
         assert telegram._tdjson.send.call_count == 0
+
+
+class TestWorkerExceptionHandling:
+    def test_worker_thread_survives_handler_exception(self):
+        import time
+        from queue import Queue
+        from telegram.worker import SimpleWorker
+
+        q = Queue()
+        worker = SimpleWorker(queue=q)
+        worker.run()
+
+        results = []
+
+        def bad_handler(update):
+            raise RuntimeError("boom")
+
+        def good_handler(update):
+            results.append(update)
+
+        q.put((bad_handler, {"@type": "test"}))
+        q.put((good_handler, {"@type": "test"}))
+
+        time.sleep(0.5)
+        worker.stop()
+
+        assert results == [{"@type": "test"}]
+
+
+class TestListenerExceptionHandling:
+    def test_listener_survives_receive_exception(self, telegram):
+        import threading
+
+        telegram._stopped = threading.Event()
+        call_count = 0
+
+        def exploding_receive():
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise RuntimeError("receive failed")
+            if call_count == 2:
+                return {"@type": "ok", "@extra": {"request_id": "test123"}}
+            telegram._stopped.set()
+            return None
+
+        telegram._tdjson.receive = exploding_receive
+        telegram._listen_to_td()
+
+        assert call_count == 3
+
+    def test_listener_exits_on_exception_when_stopped(self, telegram):
+        import threading
+
+        telegram._stopped = threading.Event()
+
+        def exploding_receive():
+            telegram._stopped.set()
+            raise RuntimeError("error during shutdown")
+
+        telegram._tdjson.receive = exploding_receive
+        telegram._listen_to_td()
+
+
+class TestRunHandlersQueueFull:
+    def test_queue_full_does_not_propagate(self, telegram):
+        def my_handler(update):
+            pass
+
+        telegram.add_update_handler("testUpdate", my_handler)
+
+        with patch.object(telegram._workers_queue, "put", side_effect=queue.Full):
+            telegram._run_handlers({"@type": "testUpdate"})
+
+
+class TestSendMessageElementError:
+    def test_raises_on_parse_error(self, telegram):
+        error_result = AsyncResult(client=telegram)
+        error_result.error = True
+        error_result.error_info = {"@type": "error", "message": "Bad HTML"}
+        error_result._ready.set()
+
+        with patch.object(telegram, "parse_text_entities", return_value=error_result):
+            with pytest.raises(RuntimeError):
+                telegram.send_message(chat_id=1, text=Spoiler("test"))
+
+    def test_raises_on_none_update(self, telegram):
+        result = AsyncResult(client=telegram)
+        result.update = None
+        result._ready.set()
+
+        with patch.object(telegram, "parse_text_entities", return_value=result):
+            with pytest.raises(RuntimeError, match="Failed to parse text entities"):
+                telegram.send_message(chat_id=1, text=Spoiler("test"))