Handler timeout leaves zombie threads (can't cancel blocking sync code) #31

if handler_task and not handler_task.done():
    handler_task.cancel()
    try:
        await asyncio.wait_for(handler_task, timeout=0.1)
    except (asyncio.CancelledError, TimeoutError):
        pass

import asyncio, threading, time, pytest
from bubus import BaseEvent, EventBus

class BlockEvent(BaseEvent[str]):
    pass

@pytest.mark.asyncio
async def test_timeout_does_not_leave_zombie_thread():
    thread_started = threading.Event()
    thread_should_stop = threading.Event()

    def _block_sync():
        thread_started.set()
        thread_should_stop.wait(timeout=30)
        return "done"

    async def blocking_handler(event: BlockEvent) -> str:
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, _block_sync)

    bus = EventBus(name="test_zombie")
    bus.on(BlockEvent, blocking_handler)
    bus._start()
    bus.dispatch(BlockEvent(event_timeout=1.0))
    try:
        await asyncio.wait_for(bus.step(timeout=1.0), timeout=5.0)
    except (asyncio.TimeoutError, TimeoutError, Exception):
        pass
    thread_should_stop.set()
    time.sleep(1)
    await bus.stop(timeout=1, clear=True)
    assert thread_started.is_set()
    zombies = [t for t in threading.enumerate()
               if t is not threading.main_thread() and t.is_alive() and not t.daemon]
    assert len(zombies) == 0, f"{len(zombies)} zombie thread(s) alive: {[t.name for t in zombies]}"