feat: upstream Perry's provider repairs (bound connect, receive shutdown, UDP datagrams, stdio, HTTP casing, HTTP/2 push/settings/ping)

crates/turnloop/src/backend/iocp/mod.rs

@@ -1467,11 +1467,56 @@ unsafe impl Backend for Iocp {
         sockopt::shutdown_read(self.socket_of(h)?)
     }
     fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> {
+        if matches!(option, SocketOption::ListenBacklog(_))
+            && self.get(h)?.transport.kind != Kind::Listener
+        {
+            return Err(invalid());
+        }
         sockopt::set(self.socket_of(h)?, option)
     }
+    fn open_bound_tcp(
+        &mut self,
+        h: Handle,
+        transport: Detached,
+        peer: SocketAddr,
+        opts: TcpOpts,
+    ) -> Result<()> {
+        if transport.kind != Kind::Tcp {
+            return Err(invalid());
+        }
+        let Native::Socket(socket) = &transport.native else {
+            return Err(invalid());
+        };
+        let raw = socket.as_raw_socket() as usize;
+        let local = socket::address(raw, false)?;
+        if local.port() == 0 || local.is_ipv4() != peer.is_ipv4() {
+            return Err(invalid());
+        }
+        match socket::address(raw, true) {
+            Ok(_) => return Err(invalid()),
+            Err(error) if error.os == Some(WSAENOTCONN) => {}
+            Err(error) => return Err(error),
+        }
+        if opts.nodelay {
+            sockopt::set(raw, SocketOption::NoDelay(true))?;
+        }
+        if peer.ip().is_loopback() {
+            socket::loopback_connect(raw);
+        }
+        self.install(h, transport, Some(peer), None)
+    }
     fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result<SocketOption> {
         sockopt::get(self.socket_of(h)?, kind)
     }
+    fn shutdown_receive(&mut self, h: Handle) -> Result<()> {
+        let transport = &self.get(h)?.transport;
+        if transport.kind != Kind::Tcp {
+            return Err(unsupported());
+        }
+        // SAFETY: live owned TCP socket. SD_RECEIVE preserves sends and does
+        // not release OVERLAPPED storage; outstanding receives still complete.
+        socket::check(unsafe { shutdown(transport.native.raw() as usize, SD_RECEIVE) })
+    }
     fn submit(&mut self, request: Request) -> Result<()> {
         if self.services.contains(request.handle) {
             return self.services.submit(request);
@@ -1872,137 +1917,7 @@ fn write_buffers(op: &Operation, mut offset: usize, out: &mut [WSABUF; MAX_IOV])
 }
 
 #[cfg(all(test, not(loom)))]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn direct_discovery_and_event_queue_draining_have_distinct_counts() {
-        use windows_sys::Win32::{
-            Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject,
-        };
-        let mut backend = Iocp::new(&Config::default(), BufferPool::new(2, 64)).expect("backend");
-        let mut events = Vec::with_capacity(1);
-        for wake in [false, true] {
-            if wake {
-                backend.port.post(WAKE, 0).expect("wake");
-            }
-            let info = backend
-                .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events)
-                .expect("discovery");
-            assert_eq!(
-                (info.waits, info.discovery_polls, info.zero_event_waits),
-                (0, 1, u32::from(!wake))
-            );
-            assert!(events.is_empty());
-        }
-        let Integration::Event(event) = backend.integration().expect("event helper") else {
-            panic!("Event required");
-        };
-        backend.port.post(WAKE, 0).expect("helper packet");
-        // SAFETY: backend owns the event throughout this bounded external wait.
-        let result = unsafe { WaitForSingleObject(event as _, 2000) };
-        assert_eq!(result, WAIT_OBJECT_0, "helper must forward a real packet");
-        let info = backend
-            .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events)
-            .expect("helper drain");
-        assert_eq!(
-            (info.waits, info.discovery_polls, info.zero_event_waits),
-            (0, 0, 0)
-        );
-        assert!(events.is_empty());
-    }
-
-    // These operations model already-dequeued kernel completions. No native I/O
-    // is submitted, so unwind must discard their metadata rather than wait on it.
-    pub(super) struct Synthetic {
-        pub backend: Iocp,
-        pub entries: [Entry; 2],
-        pub ops: [OpId; 2],
-    }
-    impl Synthetic {
-        pub fn new() -> Self {
-            let config = Config {
-                max_handles: 2,
-                max_operations: 2,
-                ..Config::default()
-            };
-            let mut backend = Iocp::new(&config, BufferPool::new(2, 64)).expect("backend");
-            let handle = Handle { owner: 1, key: 0 };
-            backend
-                .open(
-                    handle,
-                    Open::Tcp {
-                        addr: ([127, 0, 0, 1], 9).into(),
-                        opts: TcpOpts::default(),
-                    },
-                )
-                .expect("socket resource");
-            let ops = [OpId { owner: 1, key: 0 }, OpId { owner: 1, key: 1 }];
-            let entries = std::array::from_fn(|i| {
-                backend
-                    .submit(Request {
-                        op: ops[i],
-                        handle,
-                        operation: Operation::Write(WriteBuf::Owned(vec![0; i + 1])),
-                    })
-                    .expect("synthetic request");
-                let pending = backend.ops[i].as_mut().expect("op");
-                pending.waiting = true;
-                pending.queued = false;
-                pending.stage = Stage::Io;
-                Entry {
-                    key: 1,
-                    overlapped: backend.kernel_ptr(i) as usize,
-                    bytes: (i + 1) as u32,
-                    status: 0,
-                }
-            });
-            backend.ready.clear();
-            Self {
-                backend,
-                entries,
-                ops,
-            }
-        }
-        pub fn assert_delivered(&mut self) {
-            let mut out = Vec::with_capacity(1);
-            for (i, op) in self.ops.into_iter().enumerate() {
-                let info = self
-                    .backend
-                    .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut out)
-                    .expect("drain valid packet");
-                assert_eq!((info.waits, info.discovery_polls), (0, 0));
-                assert_eq!(out.len(), 1);
-                assert_eq!(out[0].op, op);
-                assert!(out[0].terminal);
-                assert!(matches!(out[0].result, Ok(Outcome::Wrote(n)) if n == i + 1));
-                out.clear();
-            }
-            assert!(self.backend.ops.iter().all(Option::is_none));
-            assert!(!self.backend.has_work());
-        }
-    }
-    impl Drop for Synthetic {
-        fn drop(&mut self) {
-            self.backend.ops.iter_mut().for_each(|op| *op = None);
-            self.backend.ready.clear();
-        }
-    }
-    #[test]
-    fn invalid_entry_does_not_discard_later_completions() {
-        let mut fixture = Synthetic::new();
-        let batch = [fixture.entries[0], Entry::default(), fixture.entries[1]];
-        assert_eq!(
-            fixture
-                .backend
-                .entries(&batch)
-                .expect_err("invalid packet")
-                .kind,
-            ErrorKind::InvalidInput
-        );
-        fixture.assert_delivered();
-    }
-}
+mod tests;
 
 /// Construct an explicitly bound socket for the portable adoption contract.
 #[cfg(test)]

crates/turnloop/src/backend/iocp/sockopt.rs

@@ -235,6 +235,18 @@ pub(super) fn set(socket: usize, option: SocketOption) -> Result<()> {
     match option {
         SocketOption::NoDelay(on) => set_int(socket, IPPROTO_TCP, TCP_NODELAY, i32::from(on)),
         SocketOption::KeepAlive(schedule) => set_keep_alive(socket, schedule),
+        SocketOption::KeepAliveEnabled(on) => {
+            set_int(socket, SOL_SOCKET, SO_KEEPALIVE, i32::from(on))
+        }
+        SocketOption::KeepAliveSettings(schedule) => set_keepalive_parameters(socket, schedule),
+        SocketOption::ListenBacklog(value) => {
+            let value = bounded(value)?;
+            if value == 0 {
+                return Err(invalid());
+            }
+            // SAFETY: the backend verified this is a live TCP listener; no ownership transfer.
+            super::socket::check(unsafe { listen(socket, value) })
+        }
         SocketOption::Linger(value) => set_linger(socket, value),
         SocketOption::RecvBufferSize(bytes) => {
             set_int(socket, SOL_SOCKET, SO_RCVBUF, bounded(bytes)?)
@@ -267,6 +279,12 @@ pub(super) fn get(socket: usize, kind: SocketOptionKind) -> Result<SocketOption>
             SocketOption::NoDelay(get_int(socket, IPPROTO_TCP, TCP_NODELAY)? != 0)
         }
         SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(socket)?),
+        SocketOptionKind::KeepAliveEnabled => {
+            SocketOption::KeepAliveEnabled(get_int(socket, SOL_SOCKET, SO_KEEPALIVE)? != 0)
+        }
+        SocketOptionKind::KeepAliveSettings => {
+            SocketOption::KeepAliveSettings(keepalive_parameters(socket)?)
+        }
         SocketOptionKind::Linger => SocketOption::Linger(get_linger(socket)?),
         SocketOptionKind::RecvBufferSize => {
             SocketOption::RecvBufferSize(get_int(socket, SOL_SOCKET, SO_RCVBUF)?.max(0) as u32)

crates/turnloop/src/backend/iocp/tests.rs

@@ -0,0 +1,127 @@
+use super::*;
+
+#[test]
+fn direct_discovery_and_event_queue_draining_have_distinct_counts() {
+    use windows_sys::Win32::{Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject};
+    let mut backend = Iocp::new(&Config::default(), BufferPool::new(2, 64)).expect("backend");
+    let mut events = Vec::with_capacity(1);
+    for wake in [false, true] {
+        if wake {
+            backend.port.post(WAKE, 0).expect("wake");
+        }
+        let info = backend
+            .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events)
+            .expect("discovery");
+        assert_eq!(
+            (info.waits, info.discovery_polls, info.zero_event_waits),
+            (0, 1, u32::from(!wake))
+        );
+        assert!(events.is_empty());
+    }
+    let Integration::Event(event) = backend.integration().expect("event helper") else {
+        panic!("Event required");
+    };
+    backend.port.post(WAKE, 0).expect("helper packet");
+    // SAFETY: backend owns the event throughout this bounded external wait.
+    let result = unsafe { WaitForSingleObject(event as _, 2000) };
+    assert_eq!(result, WAIT_OBJECT_0, "helper must forward a real packet");
+    let info = backend
+        .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events)
+        .expect("helper drain");
+    assert_eq!(
+        (info.waits, info.discovery_polls, info.zero_event_waits),
+        (0, 0, 0)
+    );
+    assert!(events.is_empty());
+}
+
+// These operations model already-dequeued kernel completions. No native I/O
+// is submitted, so unwind must discard their metadata rather than wait on it.
+pub(super) struct Synthetic {
+    pub backend: Iocp,
+    pub entries: [Entry; 2],
+    pub ops: [OpId; 2],
+}
+impl Synthetic {
+    pub fn new() -> Self {
+        let config = Config {
+            max_handles: 2,
+            max_operations: 2,
+            ..Config::default()
+        };
+        let mut backend = Iocp::new(&config, BufferPool::new(2, 64)).expect("backend");
+        let handle = Handle { owner: 1, key: 0 };
+        backend
+            .open(
+                handle,
+                Open::Tcp {
+                    addr: ([127, 0, 0, 1], 9).into(),
+                    opts: TcpOpts::default(),
+                },
+            )
+            .expect("socket resource");
+        let ops = [OpId { owner: 1, key: 0 }, OpId { owner: 1, key: 1 }];
+        let entries = std::array::from_fn(|i| {
+            backend
+                .submit(Request {
+                    op: ops[i],
+                    handle,
+                    operation: Operation::Write(WriteBuf::Owned(vec![0; i + 1])),
+                })
+                .expect("synthetic request");
+            let pending = backend.ops[i].as_mut().expect("op");
+            pending.waiting = true;
+            pending.queued = false;
+            pending.stage = Stage::Io;
+            Entry {
+                key: 1,
+                overlapped: backend.kernel_ptr(i) as usize,
+                bytes: (i + 1) as u32,
+                status: 0,
+            }
+        });
+        backend.ready.clear();
+        Self {
+            backend,
+            entries,
+            ops,
+        }
+    }
+    pub fn assert_delivered(&mut self) {
+        let mut out = Vec::with_capacity(1);
+        for (i, op) in self.ops.into_iter().enumerate() {
+            let info = self
+                .backend
+                .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut out)
+                .expect("drain valid packet");
+            assert_eq!((info.waits, info.discovery_polls), (0, 0));
+            assert_eq!(out.len(), 1);
+            assert_eq!(out[0].op, op);
+            assert!(out[0].terminal);
+            assert!(matches!(out[0].result, Ok(Outcome::Wrote(n)) if n == i + 1));
+            out.clear();
+        }
+        assert!(self.backend.ops.iter().all(Option::is_none));
+        assert!(!self.backend.has_work());
+    }
+}
+impl Drop for Synthetic {
+    fn drop(&mut self) {
+        self.backend.ops.iter_mut().for_each(|op| *op = None);
+        self.backend.ready.clear();
+    }
+}
+#[test]
+fn invalid_entry_does_not_discard_later_completions() {
+    let mut fixture = Synthetic::new();
+    let batch = [fixture.entries[0], Entry::default(), fixture.entries[1]];
+    assert_eq!(
+        fixture
+            .backend
+            .entries(&batch)
+            .expect_err("invalid packet")
+            .kind,
+        ErrorKind::InvalidInput
+    );
+    fixture.assert_delivered();
+}

crates/turnloop/src/backend/ipc.rs

@@ -90,12 +90,16 @@ pub(super) fn accept(fd: RawFd) -> Result<Detached> {
 
 pub(super) fn stdio(raw: RawFd) -> Result<Detached> {
     // SAFETY: integer fcntl command duplicates the borrowed host fd with CLOEXEC.
-    let dup = unsafe { libc::fcntl(raw, libc::F_DUPFD_CLOEXEC, 0) };
+    let dup = unsafe { libc::fcntl(raw, libc::F_DUPFD_CLOEXEC, 3) };
     if dup < 0 {
         return Err(last_error());
     }
     // SAFETY: successful fcntl returns a fresh owned descriptor.
-    classify(unsafe { OwnedFd::from_raw_fd(dup) })
+    let fd = unsafe { OwnedFd::from_raw_fd(dup) };
+    let (lease, mode) = super::stdio::Lease::acquire(raw)?;
+    let mut transport = classify(fd)?;
+    transport.share_stdio(lease, mode);
+    Ok(transport)
 }
 
 pub(super) fn classify(fd: OwnedFd) -> Result<Detached> {