Fix optional OpenSSL duplex I/O on Unix (#9173)

* Fix optional OpenSSL duplex I/O on Unix

Release connection locks before readiness waits while serializing native SSL
attempts and preserving owned write retries. Wake waiters on native progress
and socket identity invalidation, and construct callback/native exceptions
after unlocking.

Add regression coverage to the existing SSL snippet and run the optional
OpenSSL backend in the existing Unix CI matrix. Windows and AIX retain the
legacy transport pending a separate compatibility-safe driver.

Assisted-by: OpenAI Codex:model version unavailable

* Consolidate OpenSSL runtime checks into the existing CI step

Keep the minimal system and vendored OpenSSL build checks, and run the
Unix SSL regressions in the existing system OpenSSL step. Preserve the
runtime feature set and enable its macOS dependency in that job.

Assisted-by: OpenAI Codex:model version unavailable

.github/workflows/ci.yaml

@@ -275,6 +275,8 @@ jobs:
       - name: Install macOS dependencies
         if: env.RUN_RUST_TESTS == 'true'
         uses: ./.github/actions/install-macos-deps
+        with:
+          openssl: true
 
       - name: Build rust tests
         if: env.RUN_RUST_TESTS == 'true'
@@ -327,8 +329,13 @@ jobs:
           target/debug/rustpython extra_tests/snippets/stdlib_re.py
 
       - name: Test openssl build
-        if: env.RUN_RUST_TESTS == 'true' && runner.os == 'Linux'
-        run: cargo build --locked --no-default-features --features ssl-openssl
+        if: env.RUN_RUST_TESTS == 'true' && runner.os != 'Windows'
+        run: |
+          cargo build --locked --no-default-features --features ssl-openssl
+          cargo build --locked --release --features threading,jit,ssl-openssl ${{ env.CARGO_ARGS_NO_SSL }}
+          target/release/rustpython -B extra_tests/snippets/stdlib_ssl.py
+          target/release/rustpython -B -m test -v --single-process --timeout=300 test_ssl
+        timeout-minutes: 30
 
       - name: Test vendored OpenSSL build
         if: env.RUN_RUST_TESTS == 'true' && runner.os == 'Linux'

crates/stdlib/src/openssl/driver.rs

@@ -0,0 +1,264 @@
+//! Ownership rules for nonblocking OpenSSL retries.
+//!
+//! In particular, SSL_write returning WANT_WRITE forbids *any* other I/O on
+//! that SSL until the same write is retried. Its buffer must survive Python
+//! timeout, signal and nonblocking returns as well as ordinary OS waits.
+use alloc::sync::Arc;
+use core::sync::atomic::{AtomicBool, Ordering};
+
+#[derive(Clone, Debug)]
+pub(super) struct WriteOwner(Arc<AtomicBool>);
+
+impl WriteOwner {
+    pub(super) fn new() -> Self {
+        Self(Arc::new(AtomicBool::new(true)))
+    }
+
+    pub(super) fn release(&self) {
+        self.0.store(false, Ordering::Release);
+    }
+
+    fn is_active(&self) -> bool {
+        self.0.load(Ordering::Acquire)
+    }
+
+    fn is(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.0, &other.0)
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum Needs {
+    Read,
+    Write,
+}
+
+#[derive(Debug)]
+struct PendingWrite {
+    bytes: Box<[u8]>,
+    owner: WriteOwner,
+    needs: Option<Needs>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum Permit {
+    Ready,
+    Wait,
+    RetryOriginalWrite,
+    InvalidRetry,
+}
+
+#[derive(Default, Debug)]
+pub(super) struct Driver {
+    pending: Option<PendingWrite>,
+    pub(super) terminal: bool,
+}
+
+impl Driver {
+    pub(super) fn other_io(&self) -> Permit {
+        match &self.pending {
+            Some(pending) if pending.needs == Some(Needs::Write) => {
+                if pending.owner.is_active() {
+                    Permit::Wait
+                } else {
+                    Permit::RetryOriginalWrite
+                }
+            }
+            _ => Permit::Ready,
+        }
+    }
+
+    pub(super) fn write(
+        &mut self,
+        owner: &WriteOwner,
+        bytes: &mut Option<Box<[u8]>>,
+        queued: &mut bool,
+    ) -> Permit {
+        // An invocation can fail before entering native SSL. There is no
+        // continuation in that case, so its abandoned buffer is disposable.
+        if self
+            .pending
+            .as_ref()
+            .is_some_and(|pending| pending.needs.is_none() && !pending.owner.is_active())
+        {
+            self.pending = None;
+        }
+        if let Some(pending) = &mut self.pending {
+            if pending.owner.is_active() {
+                if pending.owner.is(owner) {
+                    return Permit::Ready;
+                }
+                *queued = true;
+                return Permit::Wait;
+            }
+            // A write already queued behind another call is a distinct write,
+            // even when its contents happen to match the abandoned operation.
+            if *queued || bytes.as_deref() != Some(pending.bytes.as_ref()) {
+                return Permit::InvalidRetry;
+            }
+            pending.owner = owner.clone();
+            // Retain the original allocation, not the new invocation's copy.
+            *bytes = None;
+            return Permit::Ready;
+        }
+        self.pending = Some(PendingWrite {
+            bytes: bytes.take().expect("new write owns its buffer"),
+            owner: owner.clone(),
+            needs: None,
+        });
+        Permit::Ready
+    }
+
+    pub(super) fn bytes(&self, owner: &WriteOwner) -> &[u8] {
+        let pending = self.pending.as_ref().expect("active write");
+        assert!(pending.owner.is(owner) && pending.owner.is_active());
+        &pending.bytes
+    }
+
+    pub(super) fn settle_write(&mut self, owner: &WriteOwner, needs: Option<Needs>) -> bool {
+        let pending = self.pending.as_mut().expect("active write");
+        assert!(pending.owner.is(owner) && pending.owner.is_active());
+        if let Some(needs) = needs {
+            pending.needs = Some(needs);
+        } else {
+            self.pending = None;
+        }
+        needs.is_some()
+    }
+
+    pub(super) fn fail(&mut self) {
+        self.terminal = true;
+        self.pending = None;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn payload() -> Box<[u8]> {
+        Box::from(&b"stable bytes"[..])
+    }
+
+    #[test]
+    fn want_write_is_exclusive_and_keeps_original_allocation() {
+        let mut driver = Driver::default();
+        let owner = WriteOwner::new();
+        let mut bytes = Some(payload());
+        let ptr = bytes.as_ref().unwrap().as_ptr();
+        assert_eq!(driver.write(&owner, &mut bytes, &mut false), Permit::Ready);
+        driver.settle_write(&owner, Some(Needs::Write));
+        assert_eq!(driver.other_io(), Permit::Wait);
+        assert_eq!(driver.bytes(&owner).as_ptr(), ptr);
+        owner.release();
+        assert_eq!(driver.other_io(), Permit::RetryOriginalWrite);
+        let next = WriteOwner::new();
+        assert_eq!(
+            driver.write(&next, &mut Some(payload()), &mut false),
+            Permit::Ready
+        );
+        assert_eq!(driver.bytes(&next).as_ptr(), ptr);
+        assert!(
+            !driver.settle_write(&next, None),
+            "completion disarms cleanup"
+        );
+        assert_eq!(driver.other_io(), Permit::Ready);
+    }
+
+    #[test]
+    fn want_read_allows_progress_but_not_a_different_write() {
+        let mut driver = Driver::default();
+        let owner = WriteOwner::new();
+        driver.write(&owner, &mut Some(payload()), &mut false);
+        driver.settle_write(&owner, Some(Needs::Read));
+        assert_eq!(driver.other_io(), Permit::Ready);
+        owner.release();
+        assert_eq!(driver.other_io(), Permit::Ready);
+        let other = WriteOwner::new();
+        let mut different = Some(Box::from(&b"different"[..]));
+        assert_eq!(
+            driver.write(&other, &mut different, &mut false),
+            Permit::InvalidRetry
+        );
+        assert_eq!(
+            driver.write(&other, &mut Some(payload()), &mut false),
+            Permit::Ready
+        );
+        assert_eq!(driver.bytes(&other), b"stable bytes");
+    }
+
+    #[test]
+    fn concurrent_equal_writes_are_distinct() {
+        let mut driver = Driver::default();
+        let first = WriteOwner::new();
+        let second = WriteOwner::new();
+        driver.write(&first, &mut Some(payload()), &mut false);
+        driver.settle_write(&first, Some(Needs::Write));
+        let mut bytes = Some(payload());
+        let ptr = bytes.as_ref().unwrap().as_ptr();
+        let mut queued = false;
+        assert_eq!(driver.write(&second, &mut bytes, &mut queued), Permit::Wait);
+        driver.settle_write(&first, None);
+        assert_eq!(
+            driver.write(&second, &mut bytes, &mut queued),
+            Permit::Ready
+        );
+        assert_eq!(driver.bytes(&second).as_ptr(), ptr);
+    }
+
+    #[test]
+    fn queued_writer_cannot_acknowledge_an_abandoned_write() {
+        let mut driver = Driver::default();
+        let first = WriteOwner::new();
+        let second = WriteOwner::new();
+        driver.write(&first, &mut Some(payload()), &mut false);
+        driver.settle_write(&first, Some(Needs::Read));
+        let mut bytes = Some(payload());
+        let mut queued = false;
+        assert_eq!(driver.write(&second, &mut bytes, &mut queued), Permit::Wait);
+        first.release();
+        assert_eq!(
+            driver.write(&second, &mut bytes, &mut queued),
+            Permit::InvalidRetry
+        );
+        assert!(driver.pending.is_some());
+    }
+
+    #[test]
+    fn stale_release_cannot_orphan_an_adopted_writer() {
+        let mut driver = Driver::default();
+        let first = WriteOwner::new();
+        driver.write(&first, &mut Some(payload()), &mut false);
+        assert!(driver.settle_write(&first, Some(Needs::Write)));
+        first.release();
+        let next = WriteOwner::new();
+        assert_eq!(
+            driver.write(&next, &mut Some(payload()), &mut false),
+            Permit::Ready
+        );
+        first.release();
+        assert_eq!(driver.other_io(), Permit::Wait);
+        assert_eq!(driver.bytes(&next), b"stable bytes");
+        assert!(!driver.settle_write(&next, None));
+    }
+
+    #[test]
+    fn release_before_attempt_and_terminal_error_leave_no_retry() {
+        let mut driver = Driver::default();
+        let owner = WriteOwner::new();
+        driver.write(&owner, &mut Some(payload()), &mut false);
+        owner.release();
+        let owner = WriteOwner::new();
+        let mut different = Some(Box::from(&b"new operation"[..]));
+        assert_eq!(
+            driver.write(&owner, &mut different, &mut false),
+            Permit::Ready
+        );
+        assert_eq!(driver.bytes(&owner), b"new operation");
+        driver.settle_write(&owner, Some(Needs::Write));
+        driver.fail();
+        assert!(driver.terminal);
+        owner.release();
+        assert!(driver.pending.is_none());
+    }
+}

crates/stdlib/src/openssl/waiter.rs

@@ -0,0 +1,285 @@
+use alloc::sync::{Arc, Weak};
+use rustpython_host_env::{
+    io as host_io,
+    select::{self, POLLIN, POLLNVAL, POLLOUT, PollFd},
+};
+use std::{
+    io,
+    os::fd::{AsFd, AsRawFd, OwnedFd, RawFd},
+    time::Instant,
+};
+
+/// A persistent, native-only notification for one blocked SSL operation.
+#[derive(Debug)]
+pub(super) struct ProgressWaiter {
+    read_fd: OwnedFd,
+    write_fd: OwnedFd,
+}
+
+impl ProgressWaiter {
+    pub(super) fn new() -> io::Result<Self> {
+        let (read_fd, write_fd) = wake_pipe()?;
+        Ok(Self { read_fd, write_fd })
+    }
+
+    pub(super) fn notify(&self) {
+        loop {
+            match host_io::write_once(self.write_fd.as_fd().into(), &[1]) {
+                Ok(_) => return,
+                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
+                // A full pipe already carries a persistent notification.
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => return,
+                // Both endpoints stay owned for this entire call, so neither
+                // EBADF nor EPIPE is possible without external fd corruption.
+                Err(_) => return,
+            }
+        }
+    }
+
+    /// Re-arm under the connection guard, before the final identity check and
+    /// close-observer subscription. Draining after subscribing can lose close.
+    pub(super) fn drain(&self) -> io::Result<()> {
+        let mut buffer = [0; 256];
+        loop {
+            match host_io::read_once(self.read_fd.as_fd().into(), &mut buffer) {
+                Ok(0) => {
+                    return Err(io::Error::new(
+                        io::ErrorKind::BrokenPipe,
+                        "SSL progress notification pipe closed",
+                    ));
+                }
+                Ok(_) => {}
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(()),
+                // In particular, return EINTR so the caller can check signals.
+                Err(error) => return Err(error),
+            }
+        }
+    }
+
+    /// Wait without a connection or socket guard. A missing socket means that
+    /// write ownership permits only progress notifications to resume this call.
+    /// Returns true on timeout; readiness never drains the notification pipe.
+    pub(super) fn wait(
+        &self,
+        socket_fd: Option<RawFd>,
+        write: bool,
+        deadline: Option<Instant>,
+    ) -> io::Result<bool> {
+        if socket_fd.is_some_and(|fd| fd < 0) {
+            return Err(io::Error::from_raw_os_error(libc::EBADF));
+        }
+        let mut fds = [
+            PollFd {
+                fd: self.read_fd.as_raw_fd(),
+                events: POLLIN,
+                revents: 0,
+            },
+            PollFd {
+                // poll ignores a negative fd when only progress is allowed.
+                fd: socket_fd.unwrap_or(-1),
+                // TLS consumes normal stream data, never MSG_OOB. Watching
+                // POLLPRI could spin on urgent data that this BIO cannot read.
+                events: if write { POLLOUT } else { POLLIN },
+                revents: 0,
+            },
+        ];
+        loop {
+            let (timeout, capped) = match deadline {
+                None => (-1, false),
+                Some(deadline) => {
+                    let remaining = deadline.saturating_duration_since(Instant::now());
+                    match select::duration_as_millis_ceiling(remaining) {
+                        Some(timeout) => (timeout, false),
+                        None => (i32::MAX, true),
+                    }
+                }
+            };
+            // Even an expired deadline gets a zero-time poll, as in wait_fd.
+            // EINTR is deliberately propagated for Python signal checking.
+            match select::poll_fds(&mut fds, timeout)? {
+                0 if capped => {}
+                0 => return Ok(true),
+                _ if fds.iter().any(|fd| fd.revents & POLLNVAL != 0) => {
+                    return Err(io::Error::from_raw_os_error(libc::EBADF));
+                }
+                _ => return Ok(false),
+            }
+        }
+    }
+}
+
+impl crate::socket::SocketCloseObserver for ProgressWaiter {
+    fn socket_closed(&self) {
+        self.notify();
+    }
+}
+
+fn wake_pipe() -> io::Result<(OwnedFd, OwnedFd)> {
+    cfg_select! {
+        any(
+            target_os = "dragonfly",
+            target_os = "freebsd",
+            target_os = "linux",
+            target_os = "android",
+            target_os = "netbsd",
+            target_os = "openbsd"
+        ) => {
+            rustpython_host_env::posix::pipe2(libc::O_CLOEXEC | libc::O_NONBLOCK)
+        }
+        target_os = "redox" => {
+            use std::os::fd::FromRawFd;
+
+            // Redox provides pipe2, but the host helper excludes this target.
+            let mut fds = [-1; 2];
+            // SAFETY: fds has storage for the two descriptors pipe2 writes.
+            if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) } < 0 {
+                return Err(io::Error::last_os_error());
+            }
+            // SAFETY: successful pipe2 transfers ownership of two new fds.
+            Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
+        }
+        _ => {
+            // The host helper sets CLOEXEC and owns both fds immediately, so
+            // any subsequent fcntl failure also closes both endpoints.
+            let (read_fd, write_fd) = rustpython_host_env::posix::pipe()?;
+            rustpython_host_env::fcntl::set_blocking(read_fd.as_fd(), false)?;
+            rustpython_host_env::fcntl::set_blocking(write_fd.as_fd(), false)?;
+            Ok((read_fd, write_fd))
+        }
+    }
+}
+
+/// Implementations are native-only and must not reenter the registry.
+/// Production uses only ProgressWaiter's nonblocking pipe notification.
+pub(super) trait ProgressNotify {
+    fn notify(&self);
+}
+
+impl ProgressNotify for ProgressWaiter {
+    fn notify(&self) {
+        self.notify();
+    }
+}
+
+struct Registrations<W> {
+    generation: u64,
+    waiters: Vec<Weak<W>>,
+}
+
+/// Ownership can be abandoned without acquiring the native SSL lock. Compare
+/// generations when registering so that abandonment between predicate-check
+/// and registration cannot strand a waiter.
+pub(super) struct ProgressRegistry<W = ProgressWaiter> {
+    state: std::sync::Mutex<Registrations<W>>,
+}
+
+impl<W> Default for ProgressRegistry<W> {
+    fn default() -> Self {
+        Self {
+            state: std::sync::Mutex::new(Registrations {
+                generation: 0,
+                waiters: Vec::new(),
+            }),
+        }
+    }
+}
+
+impl<W: ProgressNotify> ProgressRegistry<W> {
+    pub(super) fn generation(&self) -> u64 {
+        self.state.lock().unwrap().generation
+    }
+
+    pub(super) fn register(&self, waiter: &Arc<W>, observed: u64) {
+        let mut state = self.state.lock().unwrap();
+        let weak = Arc::downgrade(waiter);
+        state.waiters.retain(|waiter| waiter.strong_count() != 0);
+        if !state
+            .waiters
+            .iter()
+            .any(|registered| registered.ptr_eq(&weak))
+        {
+            state.waiters.push(weak);
+        }
+        // Comparison and queuing share the notify mutex. A concurrent release
+        // either changed this generation or will find this registered waiter.
+        if state.generation != observed {
+            waiter.notify();
+        }
+    }
+
+    pub(super) fn unregister(&self, waiter: &Arc<W>) {
+        let weak = Arc::downgrade(waiter);
+        self.state
+            .lock()
+            .unwrap()
+            .waiters
+            .retain(|registered| !registered.ptr_eq(&weak));
+    }
+
+    pub(super) fn notify(&self) {
+        let mut state = self.state.lock().unwrap();
+        state.generation = state.generation.wrapping_add(1);
+        state.waiters.retain(|waiter| {
+            if let Some(waiter) = waiter.upgrade() {
+                waiter.notify();
+                true
+            } else {
+                false
+            }
+        });
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use core::sync::atomic::{AtomicUsize, Ordering};
+
+    #[derive(Default)]
+    struct Counter(AtomicUsize);
+
+    impl ProgressNotify for Counter {
+        fn notify(&self) {
+            self.0.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    #[test]
+    fn release_before_registration_is_not_lost() {
+        let registry = ProgressRegistry::default();
+        let observed = registry.generation();
+        let waiter = Arc::new(Counter::default());
+        registry.notify();
+        registry.register(&waiter, observed);
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 1);
+    }
+
+    #[test]
+    fn release_after_registration_notifies_and_unregister_removes_it() {
+        let registry = ProgressRegistry::default();
+        let waiter = Arc::new(Counter::default());
+        registry.register(&waiter, registry.generation());
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 0);
+        registry.notify();
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 1);
+        registry.unregister(&waiter);
+        registry.notify();
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 1);
+    }
+
+    #[test]
+    fn unchanged_retry_does_not_self_wake_or_duplicate_registration() {
+        let registry = ProgressRegistry::default();
+        let waiter = Arc::new(Counter::default());
+        for _ in 0..10 {
+            registry.register(&waiter, registry.generation());
+        }
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 0);
+        assert_eq!(registry.state.lock().unwrap().waiters.len(), 1);
+        registry.notify();
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 1);
+        let fresh = registry.generation();
+        registry.register(&waiter, fresh);
+        assert_eq!(waiter.0.load(Ordering::Relaxed), 1);
+    }
+}

crates/stdlib/src/socket.rs

@@ -3,7 +3,14 @@
 pub(crate) use _socket::module_def;
 
 #[cfg(feature = "ssl")]
-pub(super) use _socket::{PySocket, SockWaitKind, sock_wait, timeout_error_msg};
+pub(super) use _socket::{PySocket, timeout_error_msg};
+#[cfg(any(
+    feature = "__ssl-rustls",
+    all(feature = "ssl-openssl", not(all(unix, not(target_os = "aix"))))
+))]
+pub(super) use _socket::{SockWaitKind, sock_wait};
+#[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+pub(crate) use _socket::{SocketCloseObserver, SocketIdentity};
 
 #[pymodule]
 mod _socket {
@@ -44,6 +51,10 @@ mod _socket {
         Ok(())
     }
 
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    use alloc::sync::{Arc, Weak};
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    use core::sync::atomic::AtomicBool;
     use core::sync::atomic::{AtomicI32, Ordering};
     use core::{
         mem::ManuallyDrop,
@@ -53,6 +64,8 @@ mod _socket {
     use crossbeam_utils::atomic::AtomicCell;
     use host_socket::raw::Socket;
     use num_traits::ToPrimitive;
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    use std::sync::Mutex;
     use std::{
         ffi,
         io::{self, Read, Write},
@@ -1058,6 +1071,66 @@ mod _socket {
         flags: i32,
     }
 
+    /// A native-only wakeup. Implementations must not execute Python.
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    pub(crate) trait SocketCloseObserver: Send + Sync {
+        fn socket_closed(&self);
+    }
+
+    /// Identifies one initialization, even when a later socket reuses its fd.
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    #[derive(Debug, Default)]
+    pub(crate) struct SocketIdentity {
+        closed: AtomicBool,
+        observers: Mutex<Vec<Weak<dyn SocketCloseObserver>>>,
+    }
+
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    impl SocketIdentity {
+        pub(crate) fn is_closed(&self) -> bool {
+            self.closed.load(Ordering::Acquire)
+        }
+
+        /// Register atomically with the closed check, so close-before-wait
+        /// cannot lose a wakeup. The caller retains the strong observer.
+        pub(crate) fn subscribe(&self, observer: Arc<dyn SocketCloseObserver>) -> bool {
+            let mut observers = self.observers.lock().unwrap();
+            if self.is_closed() {
+                return false;
+            }
+            observers.retain(|observer| observer.strong_count() != 0);
+            let observer = Arc::downgrade(&observer);
+            if !observers.iter().any(|current| current.ptr_eq(&observer)) {
+                observers.push(observer);
+            }
+            true
+        }
+
+        pub(crate) fn unsubscribe(&self, observer: Arc<dyn SocketCloseObserver>) {
+            let observer = Arc::downgrade(&observer);
+            self.observers
+                .lock()
+                .unwrap()
+                .retain(|current| !current.ptr_eq(&observer));
+        }
+
+        // The socket write lock is held by the caller. Return weak observers
+        // rather than invoking them while either lock is held.
+        fn invalidate(&self) -> Vec<Weak<dyn SocketCloseObserver>> {
+            let mut observers = self.observers.lock().unwrap();
+            self.closed.store(true, Ordering::Release);
+            core::mem::take(&mut *observers)
+        }
+
+        fn notify_closed(observers: Vec<Weak<dyn SocketCloseObserver>>) {
+            for observer in observers {
+                if let Some(observer) = observer.upgrade() {
+                    observer.socket_closed();
+                }
+            }
+        }
+    }
+
     #[pyattr(name = "socket")]
     #[pyattr(name = "SocketType")]
     #[pyclass(name = "socket")]
@@ -1071,6 +1144,10 @@ mod _socket {
         proto: AtomicI32,
         pub(crate) timeout: AtomicCell<f64>,
         sock: PyRwLock<Option<Socket>>,
+        // Shared access takes `sock` first, in socket -> identity-slot order.
+        // Ordinary sockets never allocate an identity or observer list.
+        #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+        ssl_identity: Mutex<Option<Arc<SocketIdentity>>>,
     }
 
     const _: () = assert!(core::mem::size_of::<Option<Socket>>() == core::mem::size_of::<Socket>());
@@ -1083,6 +1160,19 @@ mod _socket {
                 proto: AtomicI32::new(0),
                 timeout: AtomicCell::new(-1.0),
                 sock: PyRwLock::new(None),
+                #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+                ssl_identity: Mutex::new(None),
+            }
+        }
+    }
+
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    impl Drop for PySocket {
+        fn drop(&mut self) {
+            // Rust destruction must also notify when Python __del__ did not run.
+            // Keep ordinary socket destruction free of additional locking.
+            if self.ssl_identity.get_mut().unwrap().is_some() {
+                drop(self.take_sock());
             }
         }
     }
@@ -1119,6 +1209,64 @@ mod _socket {
                 .ok_or_else(|| io::Error::from_raw_os_error(CLOSED_ERR))
         }
 
+        /// Capture the current initialization under the socket read lock.
+        #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+        pub(crate) fn ssl_identity(&self) -> io::Result<Arc<SocketIdentity>> {
+            let _sock = self.sock()?;
+            let mut identity = self.ssl_identity.lock().unwrap();
+            Ok(identity
+                .get_or_insert_with(|| Arc::new(SocketIdentity::default()))
+                .clone())
+        }
+
+        /// Hold this guard only around a truly nonblocking transport syscall,
+        /// never around SSL processing, Python callbacks, or readiness waits.
+        #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+        pub(crate) fn sock_for_ssl_identity(
+            &self,
+            identity: &Arc<SocketIdentity>,
+        ) -> io::Result<PyMappedRwLockReadGuard<'_, Socket>> {
+            let sock = self.sock()?;
+            let current = self.ssl_identity.lock().unwrap();
+            if !current
+                .as_ref()
+                .is_some_and(|current| Arc::ptr_eq(current, identity) && !identity.is_closed())
+            {
+                return Err(io::Error::from_raw_os_error(CLOSED_ERR));
+            }
+            Ok(sock)
+        }
+
+        // Caller holds the socket write lock. Notifications must be delivered
+        // only after releasing that lock, including on initialization failure.
+        #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+        fn invalidate_ssl_identity_locked(&self) -> Vec<Weak<dyn SocketCloseObserver>> {
+            self.ssl_identity
+                .lock()
+                .unwrap()
+                .take()
+                .map_or_else(Vec::new, |identity| identity.invalidate())
+        }
+
+        fn take_sock(&self) -> Option<Socket> {
+            let mut guard = self.sock.write();
+            #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+            let observers = self.invalidate_ssl_identity_locked();
+            let sock = guard.take();
+            drop(guard);
+            #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+            SocketIdentity::notify_closed(observers);
+            sock
+        }
+
+        #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+        fn invalidate_failed_ssl_init(&self) {
+            let guard = self.sock.write();
+            let observers = self.invalidate_ssl_identity_locked();
+            drop(guard);
+            SocketIdentity::notify_closed(observers);
+        }
+
         /// Snapshot the inner socket without holding `self.sock` across a syscall.
         ///
         /// Blocking accept/recv/send must not keep the RwLock: `close()` needs
@@ -1171,7 +1319,18 @@ mod _socket {
             self.kind.store(masked_kind, Ordering::Relaxed);
             self.proto.store(proto, Ordering::Relaxed);
             let mut s = self.sock.write();
-            let sock = s.insert(sock);
+            #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+            let observers = self.invalidate_ssl_identity_locked();
+            let old_socket = s.replace(sock);
+            // Closing a lingering socket may block. Unlock and wake old SSL
+            // operations first, including if initialization unwinds.
+            let mut s = scopeguard::guard(s, move |s| {
+                drop(s);
+                #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+                SocketIdentity::notify_closed(observers);
+                drop(old_socket);
+            });
+            let sock = s.as_mut().unwrap();
             // If SOCK_NONBLOCK is set, use timeout 0 (non-blocking)
             #[cfg(any(
                 target_os = "android",
@@ -1202,10 +1361,13 @@ mod _socket {
             )))]
             let timeout = DEFAULT_TIMEOUT.load();
             self.timeout.store(timeout);
-            if timeout >= 0.0 {
-                sock.set_nonblocking(true)?;
-            }
-            Ok(())
+            let result = if timeout >= 0.0 {
+                sock.set_nonblocking(true)
+            } else {
+                Ok(())
+            };
+            drop(s);
+            result
         }
 
         /// returns Err(blocking)
@@ -1691,7 +1853,14 @@ mod _socket {
         type Args = SocketInitArgs;
 
         fn init(zelf: &Py<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> {
-            Self::_init(zelf, args, vm).map_err(|e| e.into_pyexception(vm))
+            let result = Self::_init(zelf, args, vm);
+            #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+            if result.is_err() {
+                // _init can fail before reaching init_inner, leaving an older
+                // raw socket in place but invalidating its SSL initialization.
+                zelf.invalidate_failed_ssl_init();
+            }
+            result.map_err(|e| e.into_pyexception(vm))
         }
     }
 
@@ -2326,7 +2495,7 @@ mod _socket {
 
         #[pymethod]
         fn close(zelf: &Py<Self>) -> io::Result<()> {
-            let sock = zelf.sock.write().take();
+            let sock = zelf.take_sock();
             if let Some(sock) = sock {
                 close_inner(into_sock_fileno(sock))?;
             }
@@ -2336,7 +2505,7 @@ mod _socket {
         #[pymethod]
         #[inline]
         fn detach(zelf: &Py<Self>) -> i64 {
-            let sock = zelf.sock.write().take();
+            let sock = zelf.take_sock();
             sock.map_or(INVALID_SOCKET as i64, |s| into_sock_fileno(s) as i64)
         }
 
@@ -3005,7 +3174,10 @@ mod _socket {
     }
 
     /// returns Ok(true) on timeout
-    #[cfg(feature = "ssl")]
+    #[cfg(any(
+        feature = "__ssl-rustls",
+        all(feature = "ssl-openssl", not(all(unix, not(target_os = "aix"))))
+    ))]
     pub(crate) fn sock_wait(
         sock: &Socket,
         wait_kind: SockWaitKind,
@@ -3650,4 +3822,112 @@ mod _socket {
         host_socket::checked_cmsg_space(length)
             .ok_or_else(|| vm.new_overflow_error("CMSG_SPACE() argument out of range"))
     }
+
+    #[cfg(test)]
+    #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))]
+    mod ssl_identity_tests {
+        use super::{PySocket, SocketCloseObserver, SocketIdentity};
+        use alloc::sync::{Arc, Weak};
+        use core::sync::atomic::{AtomicUsize, Ordering};
+
+        struct Observer {
+            identity: Weak<SocketIdentity>,
+            notifications: AtomicUsize,
+        }
+
+        impl Observer {
+            fn new(identity: &Arc<SocketIdentity>) -> Arc<Self> {
+                Arc::new(Self {
+                    identity: Arc::downgrade(identity),
+                    notifications: AtomicUsize::new(0),
+                })
+            }
+        }
+
+        impl SocketCloseObserver for Observer {
+            fn socket_closed(&self) {
+                if let Some(identity) = self.identity.upgrade() {
+                    assert!(identity.is_closed());
+                    assert!(identity.observers.try_lock().is_ok());
+                }
+                self.notifications.fetch_add(1, Ordering::Relaxed);
+            }
+        }
+
+        #[test]
+        fn subscribe_before_close_notifies_after_invalidation() {
+            let identity = Arc::new(SocketIdentity::default());
+            let observer = Observer::new(&identity);
+            assert!(identity.subscribe(observer.clone()));
+            assert!(identity.subscribe(observer.clone()));
+            let notifications = identity.invalidate();
+            assert!(identity.is_closed());
+            assert_eq!(observer.notifications.load(Ordering::Relaxed), 0);
+            assert_eq!(notifications.len(), 1);
+            SocketIdentity::notify_closed(notifications);
+            assert_eq!(observer.notifications.load(Ordering::Relaxed), 1);
+            assert!(identity.invalidate().is_empty());
+        }
+
+        #[test]
+        fn close_before_subscribe_rejects_registration() {
+            let identity = Arc::new(SocketIdentity::default());
+            SocketIdentity::notify_closed(identity.invalidate());
+            let observer = Observer::new(&identity);
+            assert!(!identity.subscribe(observer.clone()));
+            assert_eq!(observer.notifications.load(Ordering::Relaxed), 0);
+        }
+
+        #[test]
+        fn registrations_do_not_keep_observers_alive() {
+            let identity = Arc::new(SocketIdentity::default());
+            let observer = Observer::new(&identity);
+            let weak = Arc::downgrade(&observer);
+            assert!(identity.subscribe(observer.clone()));
+            drop(observer);
+            assert!(weak.upgrade().is_none());
+            let next = Observer::new(&identity);
+            assert!(identity.subscribe(next.clone()));
+            let notifications = identity.invalidate();
+            assert_eq!(notifications.len(), 1);
+            SocketIdentity::notify_closed(notifications);
+            assert_eq!(next.notifications.load(Ordering::Relaxed), 1);
+        }
+
+        #[test]
+        fn uninitialized_socket_keeps_identity_unallocated() {
+            let socket = PySocket::default();
+            assert!(socket.ssl_identity.lock().unwrap().is_none());
+            assert!(socket.ssl_identity().is_err());
+            assert!(socket.ssl_identity.lock().unwrap().is_none());
+        }
+
+        #[test]
+        fn failed_initialization_and_rust_drop_invalidate_identity() {
+            // Exercise lifecycle bookkeeping without creating an OS socket.
+            let socket = PySocket::default();
+            let old = Arc::new(SocketIdentity::default());
+            let old_observer = Observer::new(&old);
+            assert!(old.subscribe(old_observer.clone()));
+            {
+                let _socket = socket.sock.write();
+                *socket.ssl_identity.lock().unwrap() = Some(old.clone());
+            }
+            socket.invalidate_failed_ssl_init();
+            assert!(old.is_closed());
+            assert_eq!(old_observer.notifications.load(Ordering::Relaxed), 1);
+
+            let new = Arc::new(SocketIdentity::default());
+            let new_observer = Observer::new(&new);
+            assert!(new.subscribe(new_observer.clone()));
+            {
+                let _socket = socket.sock.write();
+                *socket.ssl_identity.lock().unwrap() = Some(new.clone());
+            }
+            assert!(!new.is_closed());
+            drop(socket);
+            assert!(new.is_closed());
+            assert_eq!(new_observer.notifications.load(Ordering::Relaxed), 1);
+        }
+    }
 }

extra_tests/snippets/stdlib_ssl.py

@@ -1,5 +1,10 @@
 import pathlib
+import socket
 import ssl
+import threading
+import time
+
+from test.test_ssl import HOST, ThreadedEchoServer, testing_context
 
 CERT = (
     pathlib.Path(__file__).resolve().parent.parent.parent
@@ -23,6 +28,18 @@ server_in, server_out = ssl.MemoryBIO(), ssl.MemoryBIO()
 client = client_context.wrap_bio(client_in, client_out)
 server = server_context.wrap_bio(server_in, server_out, server_side=True)
 
+# A write that starts the handshake must release its caller's buffer on WANT.
+# Save the valid retry before resizing and changing the original exporter.
+exporter = bytearray(b"saved application bytes before WANT_READ")
+saved_write = bytes(exporter)
+try:
+    client.write(exporter)
+except ssl.SSLWantReadError:
+    exporter[:] = b"changed after WANT"
+    exporter.extend(b"; no outstanding buffer export")
+else:
+    raise AssertionError("write before the peer handshake did not need a read")
+
 for _ in range(5):
     try:
         client.do_handshake()
@@ -37,6 +54,12 @@ for _ in range(5):
 client.do_handshake()
 server.do_handshake()
 
+# OpenSSL retries require the original content and length. Never retry with
+# the mutated exporter: behavior for an invalid retry is not a parity claim.
+assert client.write(saved_write) == len(saved_write)
+server_in.write(client_out.read())
+assert server.read(len(saved_write)) == saved_write
+
 try:
     server.unwrap()
 except ssl.SSLWantReadError:
@@ -44,3 +67,285 @@ except ssl.SSLWantReadError:
 client_in.write(server_out.read() + TAIL)
 client.unwrap()
 assert client_in.read() == TAIL
+
+
+TIMEOUT = 10.0
+
+
+class BoundedEchoServer(ThreadedEchoServer):
+    # Keep the canonical accept, handshake and echo implementation. Its normal
+    # context manager and handler joins need bounds for a deadlock regression.
+    class ConnectionHandler(ThreadedEchoServer.ConnectionHandler):
+        def __init__(self, server, connsock, addr):
+            super().__init__(server, connsock, addr)
+            self.sock.settimeout(TIMEOUT)
+            server.handlers.append(self)
+
+        def run(self):
+            try:
+                try:
+                    super().run()
+                finally:
+                    self.close()
+            except BaseException as exc:
+                self.server.errors.append(exc)
+
+        def join(self, timeout=TIMEOUT * 10):
+            super().join(timeout)
+            if self.is_alive():
+                self.server.errors.append(
+                    AssertionError("echo connection handler did not stop")
+                )
+                self.server.stop()
+
+    def __init__(self, context):
+        self.errors = []
+        self.handlers = []
+        super().__init__(context=context, chatty=False)
+
+    def __enter__(self):
+        assert not self._in_context
+        self._in_context = True
+        self.start(threading.Event())
+        if not self.flag.wait(TIMEOUT):
+            self.__exit__()
+            raise AssertionError("echo server did not start")
+        return self
+
+    def run(self):
+        try:
+            try:
+                super().run()
+            finally:
+                self.close()
+        except BaseException as exc:
+            self.errors.append(exc)
+
+    def __exit__(self, *args):
+        self.stop()
+        self.join(TIMEOUT * 2)
+        self._in_context = False
+        assert not self.is_alive(), "echo server did not stop"
+        for handler in self.handlers:
+            handler.join(TIMEOUT)
+            assert not handler.is_alive(), "echo connection handler did not stop"
+        if self.errors:
+            raise self.errors[0]
+        assert not self.conn_errors, self.conn_errors
+
+
+def test_write_return_during_reader_callback():
+    client_context, server_context, hostname = testing_context()
+    client_context.maximum_version = ssl.TLSVersion.TLSv1_2
+    server_context.maximum_version = ssl.TLSVersion.TLSv1_2
+    writer_thread = threading.current_thread()
+    reader_thread = None
+    round_active = False
+
+    def message_callback(conn, direction, version, content_type, msg_type, data):
+        if (
+            not round_active
+            or conn is not connection
+            or content_type != ssl._TLSContentType.HEADER
+            or msg_type != ssl._TLSContentType.APPLICATION_DATA
+        ):
+            return
+        current_thread = threading.current_thread()
+        if (
+            direction == "write"
+            and current_thread is writer_thread
+            and not write_started.is_set()
+        ):
+            write_started.set()
+            assert reader_started.wait(TIMEOUT), "reader did not start"
+            # Encourage the reader to queue behind the native writer. Python
+            # cannot force its acquisition into the native cleanup window.
+            time.sleep(0.01)
+        elif (
+            direction == "read"
+            and current_thread is reader_thread
+            and not read_callback_seen.is_set()
+        ):
+            read_callback_seen.set()
+            assert write_returned.wait(TIMEOUT), (
+                "write did not return while another thread's read callback ran"
+            )
+
+    client_context._msg_callback = message_callback
+    with BoundedEchoServer(server_context) as echo_server:
+        with client_context.wrap_socket(
+            socket.socket(), server_hostname=hostname
+        ) as connection:
+            connection.settimeout(TIMEOUT)
+            connection.connect((HOST, echo_server.port))
+            try:
+                # This is a cleanup/callback race, not another generic duplex
+                # smoke test. The canonical duplex test remains authoritative.
+                for _ in range(20):
+                    payload = b"WRITE RETURN DURING READER CALLBACK\n"
+                    write_started = threading.Event()
+                    reader_started = threading.Event()
+                    write_returned = threading.Event()
+                    read_callback_seen = threading.Event()
+                    errors = []
+
+                    def read_echo():
+                        try:
+                            assert write_started.wait(TIMEOUT), (
+                                "writer did not enter its application-record callback"
+                            )
+                            reader_started.set()
+                            received = b""
+                            while len(received) < len(payload):
+                                chunk = connection.read(len(payload) - len(received))
+                                assert chunk, (
+                                    "unexpected EOF in echoed application data"
+                                )
+                                received += chunk
+                            assert received == payload.lower(), received
+                        except BaseException as exc:
+                            errors.append(exc)
+
+                    reader_thread = threading.Thread(target=read_echo, daemon=True)
+                    round_active = True
+                    reader_thread.start()
+                    try:
+                        assert connection.write(payload) == len(payload)
+                    finally:
+                        # Set only after the Python write call has returned or
+                        # raised. The reader callback never waits on itself.
+                        write_returned.set()
+                        reader_thread.join(TIMEOUT * 2)
+                        if reader_thread.is_alive():
+                            # Interrupt transport I/O without starting SSL I/O.
+                            try:
+                                socket.socket.shutdown(connection, socket.SHUT_RDWR)
+                            except OSError:
+                                pass
+                            reader_thread.join(TIMEOUT * 2)
+                        round_active = False
+                        assert not reader_thread.is_alive(), "reader did not stop"
+                    if errors:
+                        raise errors[0]
+                    assert write_started.is_set(), "write callback was not exercised"
+                    assert read_callback_seen.is_set(), (
+                        "read callback was not exercised"
+                    )
+                connection.write(b"over\n")
+            finally:
+                round_active = False
+                client_context._msg_callback = None
+
+
+test_write_return_during_reader_callback()
+
+
+# Limit initial nonblocking writes to 4 MiB. Small socket buffers plus a peer
+# paused after its handshake must actually produce WANT_WRITE on this platform.
+WRITE_SIZE = 64 * 1024
+MAX_INITIAL_WRITES = 64
+
+
+class PausedSinkServer(BoundedEchoServer):
+    class ConnectionHandler(BoundedEchoServer.ConnectionHandler):
+        def __init__(self, server, connsock, addr):
+            super().__init__(server, connsock, addr)
+            self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
+
+        def read(self):
+            assert self.server.read_enabled.wait(TIMEOUT), "sink was not released"
+            data = super().read()
+            if not data:
+                self.server.reached_eof.set()
+            else:
+                assert data == b"x" * len(data), "pending write data changed"
+                self.server.received += len(data)
+                assert self.server.received <= (MAX_INITIAL_WRITES + 1) * WRITE_SIZE
+            return data
+
+        def write(self, data):
+            # Data was checked before canonical echo processing normalizes it.
+            # Send nothing back, avoiding two-peer application backpressure.
+            pass
+
+        def run(self):
+            try:
+                super().run()
+            finally:
+                self.server.finished.set()
+
+    def __init__(self, context):
+        self.read_enabled = threading.Event()
+        self.finished = threading.Event()
+        self.reached_eof = threading.Event()
+        self.received = 0
+        super().__init__(context)
+
+
+def test_exporter_release_after_application_want_write():
+    client_context, server_context, hostname = testing_context()
+    client_context.maximum_version = ssl.TLSVersion.TLSv1_2
+    server_context.maximum_version = ssl.TLSVersion.TLSv1_2
+    with PausedSinkServer(server_context) as sink:
+        try:
+            with client_context.wrap_socket(
+                socket.socket(), server_hostname=hostname
+            ) as connection:
+                connection.settimeout(TIMEOUT)
+                connection.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
+                connection.connect((HOST, sink.port))
+                connection.setblocking(False)
+                exporter = bytearray(b"x" * WRITE_SIZE)
+                saved_write = bytes(exporter)
+                acknowledged = 0
+                for _ in range(MAX_INITIAL_WRITES):
+                    try:
+                        acknowledged += connection.write(exporter)
+                    except ssl.SSLWantWriteError:
+                        break
+                else:
+                    raise AssertionError(
+                        "bounded application writes never reached WANT_WRITE"
+                    )
+
+                # Returning WANT must release the export while native retries
+                # retain stable original bytes. Resize and mutate only the old
+                # exporter; the retry still uses identical saved bytes/length.
+                exporter[:] = b"changed after application WANT_WRITE"
+                exporter.extend(b"; exporter is no longer held")
+                sink.read_enabled.set()
+                # A successful write may accept a prefix. The first retry
+                # must still use identical bytes/length; advance only after a
+                # positive result, and keep the same pending slice on WANT.
+                retry_deadline = time.monotonic() + TIMEOUT
+                accepted = 0
+                pending = saved_write
+                for _ in range(64):
+                    remaining = retry_deadline - time.monotonic()
+                    assert remaining > 0, "saved-byte retry deadline expired"
+                    connection.settimeout(remaining)
+                    try:
+                        count = connection.write(pending)
+                    except (ssl.SSLWantReadError, ssl.SSLWantWriteError):
+                        continue
+                    assert 0 < count <= len(pending), count
+                    accepted += count
+                    if accepted == len(saved_write):
+                        break
+                    pending = saved_write[accepted:]
+                else:
+                    raise AssertionError("saved-byte retry exceeded 64 attempts")
+                assert accepted == len(saved_write)
+
+                # TCP EOF follows every sent TLS byte. Let the sink consume to
+                # EOF before checking the total, so extra replay is detected.
+                socket.socket.shutdown(connection, socket.SHUT_WR)
+                assert sink.finished.wait(TIMEOUT), "sink did not consume through EOF"
+                assert sink.reached_eof.is_set(), "sink stopped before observing EOF"
+                assert sink.received == acknowledged + len(saved_write), sink.received
+        finally:
+            # Setup or oracle failure must not strand the paused peer.
+            sink.read_enabled.set()
+
+
+test_exporter_release_after_application_want_write()