fix(dvc-pipe-proxy): handle pre-connected Windows clients (#1447)

Author: mamoreau-devolutions

Cargo.lock

@@ -1,21 +1,25 @@
 use async_trait::async_trait;
 use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
 use tokio::net::windows::named_pipe;
+use tracing::debug;
 
 use crate::error::DvcPipeProxyError;
 use crate::os_pipe::OsPipe;
 
 const PIPE_BUFFER_SIZE: u32 = 64 * 1024;
+// ConnectNamedPipe reports this when the client wins the create/accept race.
+const ERROR_PIPE_CONNECTED: i32 = 535;
 
-/// Unix-specific implementation of the OS pipe trait.
+/// Windows-specific implementation of the OS pipe trait.
 pub(crate) struct WindowsPipe {
     pipe_server: named_pipe::NamedPipeServer,
 }
 
 #[async_trait]
 impl OsPipe for WindowsPipe {
     async fn connect(pipe_name: &str) -> Result<Self, DvcPipeProxyError> {
-        let pipe_name = format!("\\\\.\\pipe\\{pipe_name}");
+        let pipe_path = format!("\\\\.\\pipe\\{pipe_name}");
+        debug!(%pipe_name, %pipe_path, "Creating DVC proxy Windows named pipe");
 
         let pipe_server = named_pipe::ServerOptions::new()
             .first_pipe_instance(true)
@@ -25,10 +29,28 @@ impl OsPipe for WindowsPipe {
             .in_buffer_size(PIPE_BUFFER_SIZE)
             .out_buffer_size(PIPE_BUFFER_SIZE)
             .pipe_mode(named_pipe::PipeMode::Byte)
-            .create(pipe_name)
-            .map_err(DvcPipeProxyError::Io)?;
+            .create(&pipe_path)
+            .map_err(|error| {
+                debug!(%pipe_name, %pipe_path, %error, "Failed to create DVC proxy Windows named pipe");
+                DvcPipeProxyError::Io(error)
+            })?;
 
-        pipe_server.connect().await.map_err(DvcPipeProxyError::Io)?;
+        debug!(%pipe_name, %pipe_path, "Waiting for DVC proxy Windows named-pipe client");
+        match pipe_server.connect().await {
+            Ok(()) => {}
+            Err(error) if error.raw_os_error() == Some(ERROR_PIPE_CONNECTED) => {
+                debug!(
+                    %pipe_name,
+                    %pipe_path,
+                    "DVC proxy Windows named-pipe client connected before accept"
+                );
+            }
+            Err(error) => {
+                debug!(%pipe_name, %pipe_path, %error, "Failed to accept DVC proxy Windows named-pipe client");
+                return Err(DvcPipeProxyError::Io(error));
+            }
+        }
+        debug!(%pipe_name, %pipe_path, "Connected DVC proxy Windows named-pipe client");
 
         Ok(Self { pipe_server })
     }

crates/ironrdp-dvc-pipe-proxy/src/platform/windows.rs

@@ -4,7 +4,7 @@ use ironrdp_core::impl_as_any;
 use ironrdp_dvc::{DvcClientProcessor, DvcMessage, DvcProcessor};
 use ironrdp_pdu::{PduResult, pdu_other_err};
 use ironrdp_svc::SvcMessage;
-use tracing::debug;
+use tracing::{debug, error};
 
 use crate::worker::{OnWriteDvcMessage, WorkerCtx, run_worker};
 
@@ -69,17 +69,27 @@ impl DvcProcessor for DvcNamedPipeProxy {
             channel_id,
         };
 
+        #[cfg(not(target_os = "windows"))]
+        let worker = run_worker::<crate::platform::unix::UnixPipe>(ctx);
+
+        #[cfg(target_os = "windows")]
+        let worker = run_worker::<crate::platform::windows::WindowsPipe>(ctx);
+
+        if let Err(worker_error) = worker {
+            error!(
+                channel_name = %self.channel_name,
+                pipe_name = %self.named_pipe_name,
+                %worker_error,
+                "Failed to start DVC pipe proxy worker thread"
+            );
+            return Err(pdu_other_err!("start DVC pipe proxy worker: {worker_error}"));
+        }
+
         self.worker = Some(WorkerControlCtx {
             to_pipe_tx,
             abort_event,
         });
 
-        #[cfg(not(target_os = "windows"))]
-        run_worker::<crate::platform::unix::UnixPipe>(ctx);
-
-        #[cfg(target_os = "windows")]
-        run_worker::<crate::platform::windows::WindowsPipe>(ctx);
-
         Ok(vec![])
     }
 

crates/ironrdp-dvc-pipe-proxy/src/proxy.rs

@@ -1,3 +1,4 @@
+use core::time::Duration;
 use std::sync::{Arc, mpsc};
 
 use ironrdp_dvc::encode_dvc_messages;
@@ -11,6 +12,8 @@ use crate::message::RawDataDvcMessage;
 use crate::os_pipe::OsPipe;
 
 const IO_BUFFER_SIZE: usize = 1024 * 64; // 64K
+const INITIAL_RECONNECT_DELAY: Duration = Duration::from_millis(100);
+const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(5);
 
 pub(crate) type OnWriteDvcMessage = Box<dyn Fn(u32, Vec<SvcMessage>) -> PduResult<()> + Send>;
 
@@ -23,38 +26,85 @@ pub(crate) struct WorkerCtx {
     pub(crate) channel_id: u32,
 }
 
-pub(crate) fn run_worker<P: OsPipe>(ctx: WorkerCtx) {
-    let _ = std::thread::spawn(move || {
+pub(crate) fn run_worker<P: OsPipe>(ctx: WorkerCtx) -> std::io::Result<()> {
+    let thread_name = format!("ironrdp-dvc-pipe-{}", ctx.channel_id);
+    let (startup_tx, startup_rx) = mpsc::sync_channel(1);
+
+    std::thread::Builder::new().name(thread_name).spawn(move || {
         let channel_name = ctx.channel_name.clone();
         let pipe_name = ctx.pipe_name.clone();
+        debug!(%channel_name, %pipe_name, "Starting DVC pipe proxy worker thread");
 
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .map_err(DvcPipeProxyError::Io);
-
-        let runtime = match runtime {
+        let runtime = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
             Ok(runtime) => runtime,
             Err(error) => {
                 error!(
                     %channel_name,
                     %pipe_name,
-                    ?error,
-                    "DVC pipe proxy worker thread initialization failed"
+                    %error,
+                    "Failed to initialize DVC pipe proxy worker thread"
                 );
+                let _ = startup_tx.send(Err(error));
                 return;
             }
         };
 
-        if let Err(error) = runtime.block_on(worker::<P>(ctx)) {
+        let (async_tx, async_rx) = tokio::sync::mpsc::unbounded_channel();
+        let WorkerCtx {
+            on_write_dvc,
+            to_pipe_rx: std_rx,
+            abort_event,
+            pipe_name,
+            channel_name,
+            channel_id,
+        } = ctx;
+
+        let bridge_thread_name = format!("ironrdp-dvc-pipe-{channel_id}-bridge");
+        if let Err(error) = std::thread::Builder::new().name(bridge_thread_name).spawn(move || {
+            while let Ok(data) = std_rx.recv() {
+                if async_tx.send(data).is_err() {
+                    break; // Receiver dropped
+                }
+            }
+        }) {
             error!(
                 %channel_name,
                 %pipe_name,
-                ?error,
-                "DVC pipe proxy worker thread has failed"
+                %error,
+                "Failed to start DVC pipe proxy bridge thread"
             );
+            let _ = startup_tx.send(Err(error));
+            return;
+        }
+
+        let ctx = BridgedWorkerCtx {
+            on_write_dvc,
+            to_pipe_rx: async_rx,
+            abort_event,
+            pipe_name,
+            channel_name,
+            channel_id,
+        };
+
+        if startup_tx.send(Ok(())).is_err() {
+            return;
+        }
+
+        debug!(
+            channel_name = %ctx.channel_name,
+            pipe_name = %ctx.pipe_name,
+            "Started DVC pipe proxy worker thread"
+        );
+        if let Err(error) = runtime.block_on(worker::<P>(ctx)) {
+            error!(?error, "DVC pipe proxy worker thread has failed");
         }
-    });
+    })?;
+
+    startup_rx.recv().unwrap_or_else(|_| {
+        Err(std::io::Error::other(
+            "dvc pipe proxy worker stopped before startup completed",
+        ))
+    })
 }
 
 enum NextWorkerState {
@@ -134,57 +184,39 @@ async fn process_client<P: OsPipe>(ctx: &mut BridgedWorkerCtx) -> Result<NextWor
                 if let Err(error) = pipe.write_all(&data).await
                 {
                     error!(%channel_name, %pipe_name, ?error, "Failed to write to DVC pipe");
-                    continue;
+                    return Ok(NextWorkerState::Reconnect);
                 }
             }
         };
     }
 }
 
-async fn worker<P: OsPipe>(ctx: WorkerCtx) -> Result<(), DvcPipeProxyError> {
-    // Create a bridge between std::sync::mpsc and tokio for async compatibility.
-    // It is fine to use unbounded channel here because we are using it only to
-    // forward data from a bounded channel (with size IO_MPSC_CHANNEL_SIZE),
-    // so we will never have unbounded memory growth.
-    let (async_tx, async_rx) = tokio::sync::mpsc::unbounded_channel();
-
-    let WorkerCtx {
-        on_write_dvc,
-        to_pipe_rx: std_rx,
-        abort_event,
-        pipe_name,
-        channel_name,
-        channel_id,
-    } = ctx;
-
-    // Spawn a thread to bridge std::sync::mpsc to tokio::sync::mpsc.
-    std::thread::spawn(move || {
-        while let Ok(data) = std_rx.recv() {
-            if async_tx.send(data).is_err() {
-                break; // Receiver dropped
-            }
-        }
-    });
-
-    let mut bridged_ctx = BridgedWorkerCtx {
-        on_write_dvc,
-        to_pipe_rx: async_rx,
-        abort_event,
-        pipe_name,
-        channel_name,
-        channel_id,
-    };
+async fn worker<P: OsPipe>(mut bridged_ctx: BridgedWorkerCtx) -> Result<(), DvcPipeProxyError> {
+    let mut reconnect_delay = INITIAL_RECONNECT_DELAY;
+
     loop {
-        match process_client::<P>(&mut bridged_ctx).await? {
-            NextWorkerState::Abort => {
+        match process_client::<P>(&mut bridged_ctx).await {
+            Err(error) => {
+                error!(
+                    channel_name = %bridged_ctx.channel_name,
+                    pipe_name = %bridged_ctx.pipe_name,
+                    ?error,
+                    retry_delay_ms = reconnect_delay.as_millis(),
+                    "DVC pipe proxy connection failed; retrying"
+                );
+                std::thread::sleep(reconnect_delay);
+                reconnect_delay = reconnect_delay.saturating_mul(2).min(MAX_RECONNECT_DELAY);
+            }
+            Ok(NextWorkerState::Abort) => {
                 debug!(
                     channel_name = %bridged_ctx.channel_name,
                     pipe_name = %bridged_ctx.pipe_name,
                     "Abort DVC proxy worker thread"
                 );
                 break;
             }
-            NextWorkerState::Reconnect => {
+            Ok(NextWorkerState::Reconnect) => {
+                reconnect_delay = INITIAL_RECONNECT_DELAY;
                 debug!(
                     channel_name = %bridged_ctx.channel_name,
                     pipe_name = %bridged_ctx.pipe_name,

crates/ironrdp-dvc-pipe-proxy/src/worker.rs

@@ -29,6 +29,8 @@ ironrdp-async.path = "../ironrdp-async"
 ironrdp-agent = { path = "../ironrdp-agent", features = ["internal"] }
 ironrdp-client.path = "../ironrdp-client"
 ironrdp-core.path = "../ironrdp-core"
+ironrdp-dvc.path = "../ironrdp-dvc"
+ironrdp-dvc-pipe-proxy.path = "../ironrdp-dvc-pipe-proxy"
 ironrdp-input.path = "../ironrdp-input"
 ironrdp-propertyset.path = "../ironrdp-propertyset"
 ironrdp-viewer.path = "../ironrdp-viewer"
@@ -37,7 +39,7 @@ ironrdp-tls = { path = "../ironrdp-tls", features = ["rustls"] }
 semver = "1.0"
 tracing = { version = "0.1", features = ["log"] }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-tokio = { version = "1", features = ["sync", "time"] }
+tokio = { version = "1", features = ["sync", "time", "net", "rt", "macros", "io-util"] }
 uuid = { version = "1", features = ["v4"] }
 
 [lints]

crates/ironrdp-testsuite-extra/Cargo.toml

@@ -0,0 +1,44 @@
+#[cfg(windows)]
+use core::time::Duration;
+#[cfg(windows)]
+use std::sync::mpsc;
+
+#[cfg(windows)]
+use ironrdp_dvc::DvcProcessor as _;
+#[cfg(windows)]
+use ironrdp_dvc_pipe_proxy::DvcNamedPipeProxy;
+#[cfg(windows)]
+use tokio::io::AsyncWriteExt as _;
+#[cfg(windows)]
+use tokio::net::windows::named_pipe::ClientOptions;
+
+#[cfg(windows)]
+#[tokio::test]
+async fn connects_and_forwards_windows_pipe_data() {
+    let name = format!("ironrdp-dvc-pipe-proxy-test-{}", std::process::id());
+    let (callback_tx, callback_rx) = mpsc::channel();
+    let mut proxy = DvcNamedPipeProxy::new("test", &name, move |_, messages| {
+        callback_tx
+            .send(messages)
+            .expect("test callback receiver must remain alive");
+        Ok(())
+    });
+    proxy.start(1).expect("start DVC pipe proxy");
+
+    let pipe_path = format!(r"\\.\pipe\{name}");
+    let mut client = (0..200)
+        .find_map(|_| match ClientOptions::new().open(&pipe_path) {
+            Ok(client) => Some(client),
+            Err(_) => {
+                std::thread::sleep(Duration::from_millis(10));
+                None
+            }
+        })
+        .expect("DVC pipe proxy must create the pipe within two seconds");
+
+    client.write_all(b"test data").await.expect("write to DVC pipe");
+    let messages = callback_rx
+        .recv_timeout(Duration::from_secs(1))
+        .expect("DVC pipe proxy must forward pipe data to its callback");
+    assert!(!messages.is_empty(), "DVC pipe data must produce an SVC message");
+}

crates/ironrdp-testsuite-extra/tests/dvc_pipe_proxy.rs

@@ -3,4 +3,5 @@
 
 mod agent;
 mod client_config;
+mod dvc_pipe_proxy;
 mod e2e;