Report Windows signal wakeup socket errors (#9023)

Capture Winsock failures with bounded atomic storage and report them through sys.unraisablehook on the interpreter thread. Honor warn_on_full_buffer for WSAEWOULDBLOCK. Include bounded Windows socket regressions without changing Unix or CRT wakeup writes.

Assisted-by: OpenAI Codex:model identifier unavailable

crates/host_env/src/signal.rs

@@ -395,13 +395,28 @@ pub fn wakeup_fd_is_socket(fd: libc::SOCKET) -> io::Result<bool> {
     Ok(false)
 }
 
+#[cfg(windows)]
+pub struct WakeupSendError(pub i32);
+
+#[cfg(windows)]
+impl WakeupSendError {
+    pub fn is_would_block(&self) -> bool {
+        self.0 == windows_sys::Win32::Networking::WinSock::WSAEWOULDBLOCK
+    }
+
+    // Only convert to an allocating error outside the OS signal handler.
+    pub fn into_io_error(self) -> io::Error {
+        io::Error::from_raw_os_error(self.0)
+    }
+}
+
 #[cfg(windows)]
 pub fn notify_signal(
     signum: i32,
     wakeup_fd: libc::SOCKET,
     wakeup_is_socket: bool,
     sigint_event: Option<isize>,
-) {
+) -> Option<WakeupSendError> {
     if signum == libc::SIGINT
         && let Some(handle) = sigint_event
     {
@@ -411,23 +426,31 @@ pub fn notify_signal(
     }
 
     if wakeup_fd == INVALID_SOCKET {
-        return;
+        return None;
     }
 
     let sigbyte = signum as u8;
     if wakeup_is_socket {
-        unsafe {
-            let _ = windows_sys::Win32::Networking::WinSock::send(
+        let result = unsafe {
+            windows_sys::Win32::Networking::WinSock::send(
                 wakeup_fd,
                 &sigbyte as *const u8 as *const _,
                 1,
                 0,
-            );
+            )
+        };
+        if result < 0 {
+            Some(WakeupSendError(unsafe {
+                windows_sys::Win32::Networking::WinSock::WSAGetLastError()
+            }))
+        } else {
+            None
         }
     } else {
         unsafe {
             let _ = libc::write(wakeup_fd as _, &sigbyte as *const u8 as *const _, 1);
         }
+        None
     }
 }
 

crates/vm/src/signal.rs

@@ -103,6 +103,16 @@ fn trigger_signals(vm: &VirtualMachine) -> PyResult<()> {
     }
     let _guard = SignalHandlerGuard;
 
+    #[cfg(all(feature = "host_env", windows))]
+    if vm.is_main_thread() && vm.state.is_main_interpreter() {
+        // Report before a Python signal handler can raise out of the current
+        // error-reporting scope.
+        crate::stdlib::_signal::_signal::report_wakeup_send_errors(vm);
+    } else if crate::stdlib::_signal::_signal::wakeup_send_errors_pending() {
+        // A different interpreter/thread must leave the report for the main one.
+        set_triggered();
+    }
+
     let signal_handlers = vm
         .signal_handlers
         .get()

crates/vm/src/stdlib/_signal.rs

@@ -73,6 +73,13 @@ pub(crate) mod _signal {
         }
     }
 
+    #[cfg(windows)]
+    static WARN_ON_FULL_BUFFER: atomic::AtomicBool = atomic::AtomicBool::new(true);
+
+    // Bounded pending-call storage: signal handlers cannot allocate or lock.
+    #[cfg(windows)]
+    static WAKEUP_SEND_ERRORS: [atomic::AtomicI32; 32] = [const { atomic::AtomicI32::new(0) }; 32];
+
     #[cfg(any(unix, windows, target_os = "wasi"))]
     #[allow(unused_imports)]
     pub use host_signal::SIG_ERR;
@@ -344,7 +351,7 @@ pub(crate) mod _signal {
 
     #[pyfunction]
     fn set_wakeup_fd(args: SetWakeupFdArgs, vm: &VirtualMachine) -> PyResult<i64> {
-        // TODO: implement warn_on_full_buffer
+        // TODO: implement warn_on_full_buffer for the remaining wakeup write paths.
         let _ = args.warn_on_full_buffer;
         let fd = cfg_select! {
         windows => args.fd.0,
@@ -383,7 +390,10 @@ pub(crate) mod _signal {
         let old_fd = WAKEUP.swap(fd, Ordering::Relaxed);
 
         #[cfg(windows)]
-        WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed);
+        {
+            WARN_ON_FULL_BUFFER.store(args.warn_on_full_buffer, Ordering::Relaxed);
+            WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed);
+        }
 
         #[cfg(windows)]
         if old_fd == INVALID_WAKEUP {
@@ -532,14 +542,57 @@ pub(crate) mod _signal {
         signal::TRIGGERS[signum as usize].store(true, Ordering::Relaxed);
         signal::set_triggered();
 
-        host_signal::notify_signal(
+        #[cfg(not(windows))]
+        host_signal::notify_signal(signum, WAKEUP.load(Ordering::Relaxed));
+
+        #[cfg(windows)]
+        let error = host_signal::notify_signal(
             signum,
             WAKEUP.load(Ordering::Relaxed),
-            #[cfg(windows)]
             WAKEUP_IS_SOCKET.load(Ordering::Relaxed),
-            #[cfg(windows)]
             signal::get_sigint_event(),
         );
+        #[cfg(windows)]
+        if let Some(error) = error
+            && (WARN_ON_FULL_BUFFER.load(Ordering::Relaxed) || !error.is_would_block())
+        {
+            for pending in &WAKEUP_SEND_ERRORS {
+                if pending
+                    .compare_exchange(0, error.0, Ordering::Relaxed, Ordering::Relaxed)
+                    .is_ok()
+                {
+                    // The VM may have handled the signal on another thread
+                    // before the failed wakeup send completed.
+                    signal::set_triggered();
+                    break;
+                }
+            }
+        }
+    }
+
+    #[cfg(windows)]
+    pub(crate) fn wakeup_send_errors_pending() -> bool {
+        WAKEUP_SEND_ERRORS
+            .iter()
+            .any(|error| error.load(Ordering::Relaxed) != 0)
+    }
+
+    #[cfg(windows)]
+    pub(crate) fn report_wakeup_send_errors(vm: &VirtualMachine) {
+        for pending in &WAKEUP_SEND_ERRORS {
+            let error = pending.swap(0, Ordering::Relaxed);
+            if error != 0 {
+                vm.run_unraisable(
+                    host_signal::WakeupSendError(error)
+                        .into_io_error()
+                        .into_pyexception(vm),
+                    Some(
+                        "Exception ignored while trying to send to the signal wakeup fd".to_owned(),
+                    ),
+                    vm.ctx.none(),
+                );
+            }
+        }
     }
 
     /// Reset wakeup fd after fork in child process.

extra_tests/snippets/stdlib_signal.py

@@ -106,3 +106,115 @@ if "win" not in sys.platform and sys.platform != "wasi":
     signal.signal(signal.SIGALRM, signal.SIG_DFL)
     signal.signal(signal.SIGUSR1, signal.SIG_DFL)
     signal.signal(signal.SIGUSR2, signal.SIG_DFL)
+
+
+def check_windows_wakeup_reporting():
+    if sys.platform != "win32":
+        return
+
+    import errno
+    import socket
+
+    def fill_socket(write):
+        # A nonblocking Windows send can briefly fail before TCP has filled
+        # the receiving socket. Require repeated one-byte failures to settle.
+        deadline = time.monotonic() + 5.0
+        written = 0
+        blocked = 0
+        for _ in range(4096):
+            assert time.monotonic() < deadline, "socket fill timed out"
+            assert written < 16 * 1024 * 1024, "socket fill exceeded byte bound"
+            try:
+                written += write.send(b"x" * 65536)
+            except BlockingIOError:
+                try:
+                    written += write.send(b"x")
+                except BlockingIOError:
+                    blocked += 1
+                    if blocked == 4:
+                        return
+                    time.sleep(0.050)
+                    continue
+            blocked = 0
+        raise AssertionError("socket did not fill within the attempt bound")
+
+    # Use numeric loopback addresses and bounded connection/receive calls.
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
+        listener.settimeout(5.0)
+        listener.bind(("127.0.0.1", 0))
+        listener.listen(1)
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as write:
+            write.settimeout(5.0)
+            write.connect(listener.getsockname())
+            read, _ = listener.accept()
+            with read:
+                read.settimeout(5.0)
+                read.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
+                write.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
+                write.setblocking(False)
+                old_handler = signal.getsignal(signal.SIGINT)
+                old_hook = sys.unraisablehook
+                old_fd = signal.set_wakeup_fd(-1)
+                errors = []
+                delivered = []
+
+                def raise_wakeup_signal():
+                    delivered.clear()
+                    # Reinstall for each raise: Windows CRT handlers can reset.
+                    signal.signal(
+                        signal.SIGINT, lambda signum, frame: delivered.append(signum)
+                    )
+                    signal.raise_signal(signal.SIGINT)
+                    assert delivered == [signal.SIGINT]
+
+                try:
+                    sys.unraisablehook = errors.append
+                    signal.set_wakeup_fd(write.fileno())
+                    raise_wakeup_signal()
+                    assert read.recv(1) == bytes([signal.SIGINT])
+                    assert not errors
+
+                    for options in (
+                        {},
+                        {"warn_on_full_buffer": True},
+                        {"warn_on_full_buffer": False},
+                        {},
+                    ):
+                        fill_socket(write)
+                        signal.set_wakeup_fd(write.fileno(), **options)
+                        raise_wakeup_signal()
+                        if options.get("warn_on_full_buffer", True):
+                            assert len(errors) == 1, errors
+                            error = errors.pop()
+                            assert error.exc_type is BlockingIOError
+                            assert error.exc_value.winerror == errno.WSAEWOULDBLOCK
+                            assert error.exc_value.errno in (
+                                errno.EAGAIN,
+                                errno.EWOULDBLOCK,
+                            )
+                            assert error.object is None
+                            assert error.exc_traceback is error.exc_value.__traceback__
+                            assert error.err_msg == (
+                                "Exception ignored while trying to send "
+                                "to the signal wakeup fd"
+                            )
+                        else:
+                            assert not errors
+
+                    # False suppresses a full buffer, not other send failures.
+                    signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False)
+                    write.close()
+                    raise_wakeup_signal()
+                    assert len(errors) == 1, errors
+                    assert errors[0].exc_type is OSError
+                    assert errors[0].exc_value.winerror == errno.WSAENOTSOCK
+                    assert errors[0].err_msg == (
+                        "Exception ignored while trying to send to the signal wakeup fd"
+                    )
+                finally:
+                    signal.set_wakeup_fd(old_fd)
+                    signal.signal(signal.SIGINT, old_handler)
+                    sys.unraisablehook = old_hook
+
+
+check_windows_wakeup_reporting()