transport: move the connected-UDP fast-path handles into the peer module

ConnectedPeerSocket and PeerRecvDrain are node/peer-side logic: the Transport
trait never touches them, ActivePeer stores them, and node's encrypt worker
drives them. They only happened to live under transport/udp. Relocate the
handle types to a new src/peer/connected_udp/ module (socket.rs + drain.rs),
which the node handler and encrypt worker reach via node -> peer (no new edge;
a node home would have forced a peer -> node cycle).

The udp transport keeps only the kernel-construction seam: open_connected_fd,
now folded into udp/io.rs (the byte-layer home) behind a linux/macos-gated
submodule and re-exported as transport::udp::open_connected_fd. The node
handler builds the fd through it and adopts it via ConnectedPeerSocket::from_fd.

Behavior-neutral: no wire/config/metric/log change; the per-packet hot path
(bare-RawFd send_batch_gso/raw) is untouched. Preserves the Arc multi-owner
contract, the drop-drain-before-socket ordering, and the drain's detach-on-Drop
deadlock avoidance. Reconciles the old cfg(unix)/any(linux,macos) double-gate
onto the single any(linux,macos) predicate.

src/node/encrypt_worker.rs

@@ -132,8 +132,7 @@ pub(crate) struct FmpSendJob {
     /// the job completes and the worker drops it, only the peer's
     /// strong ref remains.
     #[cfg(any(target_os = "linux", target_os = "macos"))]
-    pub connected_socket:
-        Option<std::sync::Arc<crate::transport::udp::connected_peer::ConnectedPeerSocket>>,
+    pub connected_socket: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
     /// Bulk endpoint data may be dropped when the kernel reports UDP
     /// send-queue exhaustion. Control/rekey frames keep retrying so
     /// congestion cannot strand the session.
@@ -716,8 +715,7 @@ fn mac_now_ms() -> u64 {
 struct MacSequencedSendFlow {
     key: MacSendFlowKey,
     socket: AsyncUdpSocket,
-    connected_socket:
-        Option<std::sync::Arc<crate::transport::udp::connected_peer::ConnectedPeerSocket>>,
+    connected_socket: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
     dest_addr: SocketAddr,
     next_seq: std::sync::atomic::AtomicU64,
     last_used_ms: std::sync::atomic::AtomicU64,
@@ -754,9 +752,7 @@ impl MacSequencedSendFlow {
     fn spawn(
         key: MacSendFlowKey,
         socket: AsyncUdpSocket,
-        connected_socket: Option<
-            std::sync::Arc<crate::transport::udp::connected_peer::ConnectedPeerSocket>,
-        >,
+        connected_socket: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
         dest_addr: SocketAddr,
         now_ms: u64,
     ) -> Arc<Self> {
@@ -1024,8 +1020,7 @@ fn flush_batch_sync(
     struct EncryptedGroup {
         socket: AsyncUdpSocket,
         #[cfg(any(target_os = "linux", target_os = "macos"))]
-        connected_socket:
-            Option<std::sync::Arc<crate::transport::udp::connected_peer::ConnectedPeerSocket>>,
+        connected_socket: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
         dest_addr: SocketAddr,
         wire_packets: Vec<Vec<u8>>,
         drop_on_backpressure: bool,

src/node/handlers/connected_udp.rs

@@ -142,20 +142,24 @@ impl Node {
             (peer_sa, local, recv_buf, send_buf, tx)
         };
 
-        // Open the connected socket on the kernel side.
-        let socket = std::sync::Arc::new(
-            crate::transport::udp::connected_peer::ConnectedPeerSocket::open(
-                local_addr,
-                peer_socket_addr,
-                recv_buf,
-                send_buf,
-            )
-            .map_err(|e| format!("ConnectedPeerSocket::open: {e}"))?,
-        );
+        // Open the connected socket on the kernel side, then adopt the
+        // fd into the owning handle.
+        let owned = crate::transport::udp::open_connected_fd(
+            local_addr,
+            peer_socket_addr,
+            recv_buf,
+            send_buf,
+        )
+        .map_err(|e| format!("open_connected_fd: {e}"))?;
+        let socket = std::sync::Arc::new(crate::peer::connected_udp::ConnectedPeerSocket::from_fd(
+            owned,
+            peer_socket_addr,
+            local_addr,
+        ));
 
         // Spawn the drain thread. It feeds `packet_tx` exactly like
         // the wildcard listen socket — rx_loop dispatches identically.
-        let drain = crate::transport::udp::peer_drain::PeerRecvDrain::spawn(
+        let drain = crate::peer::connected_udp::PeerRecvDrain::spawn(
             socket.clone(),
             transport_id,
             peer_socket_addr,