feat(sockets, server_core): unified socket abstraction (#3321)

* refactor(server_core, server_openvr): remove alvr_restart

- Remove alvr_restart() from the C API and restart() from
  ServerCoreContext; RestartPending now handled identically to
  ShutdownPending in server_openvr
- Fix latent bug: restart path missing lifecycle_state=ShuttingDown
  and early return, causing fall-through into StartStream handshake

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix/chore(server_core, packets): misc connection_pipeline cleanups

- Internalize HandGestureManager into tracking_loop as a local variable,
  removing the Arc<Mutex<>> parameter
- Move disconnect_notif and tracking_manager resets to be grouped
  together after the handshake completes, next to bitrate_manager
- Remove redundant braces around enable_on_connect_script block
- Remove stale debug log, update log messages
- Note that server_ip in ClientConnectionResult is currently unused

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(sockets): add SocketConnection abstraction

Wraps ProtoControlSocket + StreamSocket into a unified SocketConnection
that hides the handshake protocol. Adds connect_to_client,
listen_to_server and send_restart_signal free functions.
StreamSocketConfig consolidates buffer config into a single field.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(server_core): use SocketConnection in connection_pipeline

Replace manual handshake (proto_socket.split + StreamSocketBuilder +
separate stream setup) with SocketConnection.from_client_connection.
Move connection_result recv into alvr_sockets::connect_to_client.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: zmerp

Cargo.lock

@@ -65,7 +65,7 @@ impl VideoStreamingCapabilities {
 pub struct ConnectionAcceptedInfo {
     pub client_protocol_id: u64,
     pub platform_string: String,
-    pub server_ip: IpAddr,
+    pub server_ip: IpAddr, // must be unused for now
     pub streaming_capabilities: Option<VideoStreamingCapabilities>,
 }
 

alvr/packets/src/lib.rs

@@ -562,13 +562,6 @@ pub unsafe extern "C" fn alvr_duration_until_next_vsync(out_ns: *mut u64) -> boo
     }
 }
 
-#[unsafe(no_mangle)]
-pub extern "C" fn alvr_restart() {
-    if let Some(context) = SERVER_CORE_CONTEXT.write().take() {
-        context.restart();
-    }
-}
-
 #[unsafe(no_mangle)]
 pub extern "C" fn alvr_shutdown() {
     SERVER_CORE_CONTEXT.write().take();

alvr/server_core/src/c_api.rs

@@ -2,7 +2,6 @@ use crate::{
     ConnectionContext, FILESYSTEM_LAYOUT, SESSION_MANAGER, ServerCoreEvent,
     ServerNegotiatedStreamingConfig,
     bitrate::BitrateManager,
-    hand_gestures::HandGestureManager,
     input_mapping::ButtonMappingManager,
     sockets::WelcomeSocket,
     statistics::StatisticsManager,
@@ -30,8 +29,8 @@ use alvr_session::{
     SocketProtocol, SteamvrHmdInitConfig,
 };
 use alvr_sockets::{
-    CONTROL_PORT, KEEPALIVE_INTERVAL, KEEPALIVE_TIMEOUT, PeerType, ProtoControlSocket,
-    StreamSocketBuilder, WIRED_CLIENT_HOSTNAME,
+    CONTROL_PORT, KEEPALIVE_INTERVAL, KEEPALIVE_TIMEOUT, ProtoControlSocket, SocketConnection,
+    StreamSocketConfig, WIRED_CLIENT_HOSTNAME,
 };
 use std::{
     collections::{HashMap, hash_map::DefaultHasher},
@@ -485,9 +484,9 @@ fn try_connect(
 ) -> ConResult {
     dbg_connection!("try_connect: Finding client and creating control socket");
 
-    let (proto_socket, client_ip) = ProtoControlSocket::connect_to(
+    let (socket, client_ip, connection_result) = alvr_sockets::connect_to_client(
+        client_ips.keys().cloned().collect(),
         Duration::from_secs(1),
-        PeerType::AnyClient(client_ips.keys().cloned().collect()),
     )?;
 
     let Some(client_hostname) = client_ips.remove(&client_ip) else {
@@ -502,7 +501,8 @@ fn try_connect(
             if let Err(e) = connection_pipeline(
                 Arc::clone(&ctx),
                 lifecycle_state,
-                proto_socket,
+                socket,
+                connection_result,
                 client_hostname.clone(),
                 client_ip,
             ) {
@@ -530,7 +530,8 @@ fn try_connect(
 fn connection_pipeline(
     ctx: Arc<ConnectionContext>,
     lifecycle_state: Arc<RwLock<LifecycleState>>,
-    mut proto_socket: ProtoControlSocket,
+    socket: ProtoControlSocket,
+    connection_result: ClientConnectionResult,
     client_hostname: String,
     client_ip: IpAddr,
 ) -> ConResult {
@@ -551,21 +552,6 @@ fn connection_pipeline(
         ClientConnectionsAction::UpdateCurrentIp(Some(client_ip)),
     );
 
-    let disconnect_notif = Arc::new(Condvar::new());
-
-    dbg_connection!("connection_pipeline: Getting client status packet");
-    let connection_result = match proto_socket.recv(HANDSHAKE_ACTION_TIMEOUT) {
-        Ok(r) => r,
-        Err(ConnectionError::TryAgain(e)) => {
-            debug!(
-                "Failed to recive client connection packet. This is normal for USB connection.\n{e}"
-            );
-
-            return Ok(());
-        }
-        Err(e) => return Err(e),
-    };
-
     let maybe_streaming_caps =
         if let ClientConnectionResult::ConnectionAccepted(info) = connection_result {
             session_manager_lock.update_client_connections(
@@ -593,8 +579,6 @@ fn connection_pipeline(
         con_bail!("Only streaming clients are supported for now");
     };
 
-    dbg_connection!("connection_pipeline: setting up negotiated streaming config");
-
     let initial_settings = session_manager_lock.settings().clone();
 
     fn get_view_res(config: FrameSize, default_res: UVec2) -> UVec2 {
@@ -801,10 +785,6 @@ fn connection_pipeline(
         .with_ext(NegotiatedStreamingConfigExt {}),
     )
     .to_con()?;
-    proto_socket.send(&stream_config_packet).to_con()?;
-
-    let (mut control_sender, mut control_receiver) =
-        proto_socket.split(STREAMING_RECV_TIMEOUT).to_con()?;
 
     let new_steamvr_hmd_init_config = SteamvrHmdInitConfig {
         eye_resolution_width: transcoding_view_resolution.x,
@@ -819,21 +799,38 @@ fn connection_pipeline(
         session.steamvr_hmd_init_config = new_steamvr_hmd_init_config;
         session.restart_settings_hash = new_hash;
 
-        control_sender.send(&ServerControlPacket::Restarting).ok();
+        alvr_sockets::send_restart_signal(socket, stream_config_packet)?;
 
         crate::notify_restart_driver();
-    }
 
-    dbg_connection!("connection_pipeline: Send StartStream packet");
-    control_sender
-        .send(&ServerControlPacket::StartStream)
-        .to_con()?;
+        *lifecycle_state.write() = LifecycleState::ShuttingDown;
 
-    let signal = control_receiver.recv(HANDSHAKE_ACTION_TIMEOUT)?;
-    if !matches!(signal, ClientControlPacket::StreamReady) {
-        con_bail!("Got unexpected packet waiting for stream ack");
+        return Ok(());
     }
-    dbg_connection!("connection_pipeline: Got StreamReady packet");
+
+    let stream_protocol = if wired {
+        SocketProtocol::Tcp
+    } else {
+        initial_settings.connection.stream_protocol
+    };
+
+    dbg_connection!("connection_pipeline: Finishing handshake");
+    let mut socket = SocketConnection::from_client_connection(
+        socket,
+        HANDSHAKE_ACTION_TIMEOUT,
+        stream_config_packet,
+        StreamSocketConfig {
+            protocol: stream_protocol,
+            port: initial_settings.connection.stream_port,
+            buffer_config: initial_settings.connection.server_buffer_config,
+            max_packet_size: initial_settings.connection.packet_size as _,
+            dscp: initial_settings.connection.dscp,
+        },
+    )?;
+
+    dbg_connection!("connection_pipeline: Handshake successful, spawning threads");
+
+    let disconnect_notif = Arc::new(Condvar::new());
 
     *ctx.statistics_manager.write() = Some(StatisticsManager::new(
         initial_settings.connection.statistics_history_size,
@@ -844,36 +841,23 @@ fn connection_pipeline(
             0.0
         },
     ));
-
     *ctx.bitrate_manager.lock() =
         BitrateManager::new(initial_settings.video.bitrate.history_size, fps);
+    *ctx.tracking_manager.write() =
+        TrackingManager::new(initial_settings.connection.statistics_history_size);
 
-    let stream_protocol = if wired {
-        SocketProtocol::Tcp
-    } else {
-        initial_settings.connection.stream_protocol
-    };
-
-    dbg_connection!("connection_pipeline: StreamSocket connect_to_client");
-    let mut stream_socket = StreamSocketBuilder::connect_to_client(
-        HANDSHAKE_ACTION_TIMEOUT,
-        client_ip,
-        initial_settings.connection.stream_port,
-        stream_protocol,
-        initial_settings.connection.dscp,
-        initial_settings.connection.server_buffer_config,
-        initial_settings.connection.packet_size as _,
-    )?;
+    let control_sender = Arc::new(Mutex::new(socket.request_reliable_stream()?));
+    let mut video_sender = socket.request_unreliable_stream(VIDEO);
+    let game_audio_sender: alvr_sockets::StreamSender<()> = socket.request_unreliable_stream(AUDIO);
+    let haptics_sender = socket.request_unreliable_stream(HAPTICS);
 
-    let mut video_sender = stream_socket.request_stream(VIDEO);
-    let game_audio_sender: alvr_sockets::StreamSender<()> = stream_socket.request_stream(AUDIO);
+    let mut control_receiver = socket.subscribe_to_reliable_stream()?;
     let mut microphone_receiver: alvr_sockets::StreamReceiver<()> =
-        stream_socket.subscribe_to_stream(AUDIO, MAX_UNREAD_PACKETS);
+        socket.subscribe_to_unreliable_stream(AUDIO, MAX_UNREAD_PACKETS);
     let tracking_receiver =
-        stream_socket.subscribe_to_stream::<TrackingData>(TRACKING, MAX_UNREAD_PACKETS);
-    let haptics_sender = stream_socket.request_stream(HAPTICS);
+        socket.subscribe_to_unreliable_stream::<TrackingData>(TRACKING, MAX_UNREAD_PACKETS);
     let mut statics_receiver =
-        stream_socket.subscribe_to_stream::<ClientStatistics>(STATISTICS, MAX_UNREAD_PACKETS);
+        socket.subscribe_to_unreliable_stream::<ClientStatistics>(STATISTICS, MAX_UNREAD_PACKETS);
 
     let (video_channel_sender, video_channel_receiver) =
         std::sync::mpsc::sync_channel(initial_settings.connection.max_queued_server_video_frames);
@@ -906,7 +890,7 @@ fn connection_pipeline(
         }
     });
 
-    #[cfg_attr(target_os = "linux", allow(unused_variables))]
+    #[cfg_attr(target_os = "linux", expect(unused_variables))]
     let game_audio_thread = if let Switch::Enabled(config) =
         initial_settings.audio.game_audio.clone()
     {
@@ -1059,23 +1043,14 @@ fn connection_pipeline(
         }
     };
 
-    *ctx.tracking_manager.write() =
-        TrackingManager::new(initial_settings.connection.statistics_history_size);
-    let hand_gesture_manager = Arc::new(Mutex::new(HandGestureManager::new()));
-
     let tracking_receive_thread = thread::spawn({
         let ctx = Arc::clone(&ctx);
-        let hand_gesture_manager = Arc::clone(&hand_gesture_manager);
         let initial_settings = initial_settings.clone();
         let client_hostname = client_hostname.clone();
         move || {
-            tracking::tracking_loop(
-                &ctx,
-                initial_settings,
-                hand_gesture_manager,
-                tracking_receiver,
-                || is_streaming(&client_hostname),
-            );
+            tracking::tracking_loop(&ctx, initial_settings, tracking_receiver, || {
+                is_streaming(&client_hostname)
+            });
         }
     });
 
@@ -1114,8 +1089,6 @@ fn connection_pipeline(
         }
     });
 
-    let control_sender = Arc::new(Mutex::new(control_sender));
-
     let real_time_update_thread = thread::spawn({
         let control_sender = Arc::clone(&control_sender);
         let client_hostname = client_hostname.clone();
@@ -1341,7 +1314,7 @@ fn connection_pipeline(
         let client_hostname = client_hostname.clone();
         move || {
             while is_streaming(&client_hostname) {
-                match stream_socket.recv() {
+                match socket.recv_poll() {
                     Ok(()) => (),
                     Err(ConnectionError::TryAgain(_)) => continue,
                     Err(e) => {
@@ -1374,22 +1347,19 @@ fn connection_pipeline(
         }
     });
 
-    {
-        if initial_settings.connection.enable_on_connect_script {
-            let on_connect_script = FILESYSTEM_LAYOUT.get().map(|l| l.connect_script()).unwrap();
-            info!(
-                "Running on connect script (connect): {}",
-                on_connect_script.display()
-            );
-            if let Err(e) = Command::new(&on_connect_script)
-                .env("ACTION", "connect")
-                .spawn()
-            {
-                warn!("Failed to run connect script: {e}");
-            }
+    if initial_settings.connection.enable_on_connect_script {
+        let on_connect_script = FILESYSTEM_LAYOUT.get().map(|l| l.connect_script()).unwrap();
+        info!(
+            "Running on connect script (connect): {}",
+            on_connect_script.display()
+        );
+        if let Err(e) = Command::new(&on_connect_script)
+            .env("ACTION", "connect")
+            .spawn()
+        {
+            warn!("Failed to run connect script: {e}");
         }
     }
-
     if initial_settings.extra.capture.startup_video_recording {
         info!("Creating recording file");
         crate::create_recording_file(&ctx, session_manager_lock.settings());
@@ -1416,7 +1386,7 @@ fn connection_pipeline(
         ))
         .ok();
 
-    dbg_connection!("connection_pipeline: handshake finished; unlocking streams");
+    dbg_connection!("connection_pipeline: Threads initialized; unlocking streams");
     alvr_common::wait_rwlock(&disconnect_notif, &mut session_manager_lock);
     dbg_connection!("connection_pipeline: Begin connection shutdown");
 

alvr/server_core/src/connection.rs

@@ -16,8 +16,8 @@ pub use tracking::HandType;
 
 use crate::connection::VideoPacket;
 use alvr_common::{
-    ConnectionState, DEVICE_ID_TO_PATH, DeviceMotion, LifecycleState, Pose, RelaxedAtomic,
-    ViewParams, dbg_server_core, error,
+    ConnectionState, DEVICE_ID_TO_PATH, DeviceMotion, LifecycleState, Pose, ViewParams,
+    dbg_server_core, error,
     glam::{UVec2, Vec2},
     parking_lot::{Mutex, RwLock},
     settings_schema::Switch,
@@ -185,7 +185,6 @@ pub fn registered_button_set() -> HashSet<u64> {
 
 pub struct ServerCoreContext {
     lifecycle_state: Arc<RwLock<LifecycleState>>,
-    is_restarting: RelaxedAtomic,
     connection_context: Arc<ConnectionContext>,
     connection_thread: Arc<RwLock<Option<JoinHandle<()>>>>,
     webserver_runtime: Option<Runtime>,
@@ -246,9 +245,7 @@ impl ServerCoreContext {
         (
             Self {
                 lifecycle_state: Arc::new(RwLock::new(LifecycleState::StartingUp)),
-                is_restarting: RelaxedAtomic::new(false),
                 connection_context,
-
                 connection_thread: Arc::new(RwLock::new(None)),
                 webserver_runtime: Some(webserver_runtime),
             },
@@ -530,14 +527,6 @@ impl ServerCoreContext {
             .as_mut()
             .map(|stats| stats.duration_until_next_vsync())
     }
-
-    pub fn restart(self) {
-        dbg_server_core!("restart");
-
-        self.is_restarting.set(true);
-
-        // drop is called here for self
-    }
 }
 
 impl Drop for ServerCoreContext {