fix for windows tests

Author: griffinmilsap

src/ezmsg/core/backend.py

@@ -462,13 +462,16 @@ def _shutdown_was_unclean(self) -> bool:
         return main_shutdown_errors or loop_unclean
 
     def _exit_with_sigint(self) -> None:
+        code = 0xC000013A if os.name == "nt" else 130
+        if os.name == "nt":
+            os._exit(code)
+
         prev_handler = None
         try:
             prev_handler = signal.getsignal(signal.SIGINT)
             signal.signal(signal.SIGINT, signal.SIG_DFL)
             signal.raise_signal(signal.SIGINT)
         except Exception:
-            code = 0xC000013A if os.name == "nt" else 130
             raise SystemExit(code)
         finally:
             if prev_handler is not None:
@@ -477,7 +480,6 @@ def _exit_with_sigint(self) -> None:
                 except Exception:
                     pass
 
-        code = 0xC000013A if os.name == "nt" else 130
         raise SystemExit(code)
 
     def _cleanup(self) -> None:

src/ezmsg/core/backendprocess.py

@@ -44,13 +44,15 @@ def _strict_shutdown_enabled() -> bool:
 class ShutdownSummary:
     cancelled_tasks: int = 0
     executor_active: int = 0
+    pending_tasks: int = 0
     suppressed_errors: int = 0
     forced_interrupt: bool = False
 
     @property
     def unclean(self) -> bool:
         return bool(
             self.executor_active
+            or self.pending_tasks
             or self.suppressed_errors
             or self.forced_interrupt
         )
@@ -613,36 +615,45 @@ def _loop_exception_handler(
         if not strict_shutdown:
             shutdown_suppress.set()
             # Cancel and await remaining tasks before stopping the loop.
-            async def _cancel_remaining() -> int:
+            async def _cancel_remaining(timeout: float = 1.0) -> tuple[int, int]:
                 tasks = [
                     t
                     for t in asyncio.all_tasks()
                     if t is not asyncio.current_task() and not t.done()
                 ]
                 for t in tasks:
                     t.cancel()
-                if tasks:
-                    await asyncio.wait(tasks)
-                return len(tasks)
+                if not tasks:
+                    return 0, 0
+                _, pending = await asyncio.wait(tasks, timeout=timeout)
+                return len(tasks), len(pending)
 
             cancelled_count = 0
+            pending_count = 0
             forced_interrupt = False
             fut = asyncio.run_coroutine_threadsafe(_cancel_remaining(), loop)
             try:
-                cancelled_count = fut.result()
+                cancelled_count, pending_count = fut.result()
             except KeyboardInterrupt:
                 forced_interrupt = True
                 fut.cancel()
             except Exception:
                 cancelled_count = 0
+                pending_count = 0
 
             suppressed_count = suppressed_shutdown_errors["count"]
-            if cancelled_count or suppressed_count or forced_interrupt:
+            if cancelled_count or suppressed_count or forced_interrupt or pending_count:
                 if forced_interrupt and not cancelled_count and not suppressed_count:
                     logger.warning(
                         "Shutdown interrupted; tasks may still be running. "
                         "Re-run with EZMSG_STRICT_SHUTDOWN=1 to debug tasks with poor shutdown behavior."
                     )
+                elif pending_count:
+                    logger.warning(
+                        "Shutdown timed out waiting for %d task(s). "
+                        "Re-run with EZMSG_STRICT_SHUTDOWN=1 to debug tasks with poor shutdown behavior.",
+                        pending_count,
+                    )
                 else:
                     logger.warning(
                         "Shutdown suppressed %d error(s) and cancelled %d task(s). "
@@ -654,6 +665,7 @@ async def _cancel_remaining() -> int:
 
             if shutdown_summary is not None:
                 shutdown_summary.cancelled_tasks = cancelled_count
+                shutdown_summary.pending_tasks = pending_count
                 shutdown_summary.executor_active = (
                     executor.active_count() if executor is not None else 0
                 )