Fix "complete()" possibly hanging forever if "enqueue=True" (Delgan#647)

CHANGELOG.rst

@@ -5,6 +5,7 @@
 - Make ``patch()`` calls cumulative instead of overriding the possibly existing patching function (`#462 <https://github.com/Delgan/loguru/issues/462>`_).
 - Fix logs colorization not automatically enabled for Jupyter Notebook and Google Colab (`#494 <https://github.com/Delgan/loguru/issues/494>`_).
 - Fix logs colorization not automatically enabled for Github Actions and others CI platforms (`#604 <https://github.com/Delgan/loguru/issues/604>`_).
+- Fix ``logger.complete()`` possibly hanging forever when ``enqueue=True`` and ``catch=False`` if internal thread killed due to ``Exception`` raised by sink (`#647 <https://github.com/Delgan/loguru/issues/647>`_).
 - Raise exception if ``logger.catch()`` is used to wrap a class instead of a function to avoid unexpected behavior (`#623 <https://github.com/Delgan/loguru/issues/623>`_).
 
 

loguru/_handler.py

@@ -273,6 +273,7 @@ def _queued_writer(self):
             except Exception:
                 with lock:
                     if not self._error_interceptor.should_catch():
+                        self._confirmation_event.set()
                         raise
                     self._error_interceptor.print(None)
                 continue
@@ -289,6 +290,7 @@ def _queued_writer(self):
                     self._sink.write(message)
                 except Exception:
                     if not self._error_interceptor.should_catch():
+                        self._confirmation_event.set()
                         raise
                     self._error_interceptor.print(message.record)
 

tests/test_add_option_enqueue.py

@@ -4,7 +4,6 @@
 import time
 
 import pytest
-
 from loguru import logger
 
 from .conftest import default_threading_excepthook
@@ -173,6 +172,37 @@ def test_not_caught_exception_sink_write(capsys):
     assert lines[-1] == "RuntimeError: You asked me to fail..."
 
 
+def test_not_caught_exception_sink_write_then_complete(capsys):
+    logger.add(NotWritable(), enqueue=True, catch=False, format="{message}")
+
+    with default_threading_excepthook():
+        logger.bind(fail=True).info("Bye bye...")
+        logger.complete()
+        logger.remove()
+
+    out, err = capsys.readouterr()
+    lines = err.strip().splitlines()
+    assert out == ""
+    assert lines[0].startswith("Exception")
+    assert lines[-1] == "RuntimeError: You asked me to fail..."
+
+
+def test_not_caught_exception_queue_get_then_complete(writer, capsys):
+    logger.add(writer, enqueue=True, catch=False, format="{message}")
+
+    with default_threading_excepthook():
+        logger.bind(broken=NotUnpicklable()).info("Bye bye...")
+        logger.complete()
+        logger.remove()
+
+    out, err = capsys.readouterr()
+    lines = err.strip().splitlines()
+    assert writer.read() == ""
+    assert out == ""
+    assert lines[0].startswith("Exception")
+    assert lines[-1].endswith("UnpicklingError: You shall not de-serialize me!")
+
+
 def test_wait_for_all_messages_enqueued(capsys):
     def slow_sink(message):
         time.sleep(0.01)