Add more robust RPC handling + tests

src/windows/wslopenvmm/CMakeLists.txt

@@ -26,7 +26,7 @@ add_custom_command(
     OUTPUT ${WSLOPENVMM_DLL}
     COMMAND ${CMAKE_COMMAND} -E env "VM_SERVICE_PROTO=${VM_SERVICE_PROTO}" ${CARGO_EXECUTABLE} build --manifest-path ${CMAKE_CURRENT_LIST_DIR}/Cargo.toml --target ${RUST_TARGET} --target-dir ${CARGO_TARGET_DIR} ${CARGO_BUILD_ARGUMENTS}
     COMMAND ${CMAKE_COMMAND} -E copy_if_different ${CARGO_OUTPUT} ${WSLOPENVMM_DLL}
-    DEPENDS ${VM_SERVICE_PROTO} Cargo.lock Cargo.toml build.rs src/client.rs src/lib.rs src/af_unix.rs wslopenvmm.h
+    DEPENDS ${VM_SERVICE_PROTO} Cargo.lock Cargo.toml build.rs src/client.rs src/lib.rs src/af_unix.rs src/rpc.rs wslopenvmm.h
     WORKING_DIRECTORY ${CMAKE_CURRENT_LIST_DIR}
     VERBATIM)
 

src/windows/wslopenvmm/Cargo.toml

@@ -16,7 +16,7 @@ prost = "0.14"
 prost-types = "0.14"
 tonic = "0.14"
 tonic-prost = "0.14"
-tokio = { version = "1", features = ["net", "rt", "time"] }
+tokio = { version = "1", features = ["net", "rt", "time", "macros", "sync"] }
 tower = "0.5"
 windows = { version = "0.62", features = ["Win32_Foundation"] }
 windows-sys = { version = "0.61", features = ["Win32_Networking_WinSock"] }

src/windows/wslopenvmm/src/af_unix.rs

@@ -1,5 +1,6 @@
 // Copyright (C) Microsoft Corporation. All rights reserved.
 
+use std::error::Error;
 use std::io;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
@@ -14,14 +15,33 @@ use tower::service_fn;
 pub async fn connect_channel(
     socket_path: PathBuf,
     timeout: Duration,
-) -> Result<Channel, tonic::transport::Error> {
-    Endpoint::from_static("http://[::]:50051")
-        .connect_timeout(timeout)
-        .connect_with_connector(service_fn(move |_| {
-            let socket_path = socket_path.clone();
-            async move { open_socket(&socket_path, timeout).await.map(TokioIo::new) }
-        }))
+) -> Result<Channel, io::Error> {
+    let endpoint = Endpoint::from_static("http://[::]:50051").connect_timeout(timeout);
+    let connection = endpoint.connect_with_connector(service_fn(move |_| {
+        let socket_path = socket_path.clone();
+        async move { open_socket(&socket_path, timeout).await.map(TokioIo::new) }
+    }));
+    tokio::time::timeout(timeout, connection)
         .await
+        .map_err(|_| {
+            io::Error::new(
+                io::ErrorKind::TimedOut,
+                "OpenVMM connection deadline expired",
+            )
+        })?
+        .map_err(|error| {
+            let mut source = error.source();
+            while let Some(cause) = source {
+                if let Some(error) = cause.downcast_ref::<io::Error>() {
+                    return match error.raw_os_error() {
+                        Some(code) => io::Error::from_raw_os_error(code),
+                        None => io::Error::new(error.kind(), error.to_string()),
+                    };
+                }
+                source = cause.source();
+            }
+            io::Error::new(io::ErrorKind::ConnectionAborted, error)
+        })
 }
 
 async fn open_socket(path: &Path, timeout: Duration) -> io::Result<TcpStream> {
@@ -62,4 +82,4 @@ async fn connect_socket(address: &SockAddr) -> io::Result<TcpStream> {
         return Err(error);
     }
     Ok(stream)
-}
+}

src/windows/wslopenvmm/src/client.rs

@@ -1,46 +1,101 @@
 // Copyright (C) Microsoft Corporation. All rights reserved.
 
 use std::collections::BTreeMap;
-use std::time::Duration;
+use std::future::Future;
+use std::time::{Duration, Instant};
 
+use parking_lot::Mutex;
 use tokio::runtime::Runtime;
+use tokio::sync::watch;
 use tonic::Request;
 use vmservice::{
     ConsommeBackend, CreateVmRequest, DirectBoot, DiskType, ModifyResourceRequest, ModifyType,
     NicConfig, PortConfig, ScsiDisk, VirtioConsoleConfig, VmConfig, vm_client::VmClient,
 };
-use windows::Win32::Foundation::{E_FAIL, E_INVALIDARG, ERROR_ALREADY_EXISTS, ERROR_CONNECTION_ABORTED, ERROR_NOT_FOUND, S_OK, WAIT_TIMEOUT};
+use windows::Win32::Foundation::{
+    E_FAIL, E_INVALIDARG, ERROR_ALREADY_EXISTS, ERROR_NOT_FOUND, S_OK,
+};
 use windows::core::{GUID, HRESULT};
 use windows_sys::Win32::Networking::WinSock::{AF_INET, AF_INET6};
 
-use crate::{af_unix, vmservice};
+use crate::{af_unix, rpc, vmservice};
 
 #[cfg(test)]
 mod tests;
 
+#[derive(Debug, PartialEq, Eq)]
+enum VmConfigStatus {
+    Ready,
+    CreationOutcomeUnknown,
+}
+
 pub struct VmConfigBuilder {
-    inner: VmConfig,
+    config: VmConfig,
+    status: VmConfigStatus,
+}
+
+pub struct VmConfigHandle {
+    builder: Mutex<VmConfigBuilder>,
+}
+
+impl VmConfigHandle {
+    pub fn new() -> Self {
+        Self {
+            builder: Mutex::new(VmConfigBuilder::new()),
+        }
+    }
+
+    pub fn update(&self, operation: impl FnOnce(&mut VmConfigBuilder) -> HRESULT) -> HRESULT {
+        operation(&mut self.builder.lock())
+    }
+
+    pub fn create_vm(&self, socket_path: String, timeout_ms: u32) -> Result<VmHandle, HRESULT> {
+        if timeout_ms == 0 {
+            return Err(E_INVALIDARG);
+        }
+        let timeout = Duration::from_millis(u64::from(timeout_ms));
+        let deadline = Instant::now() + timeout;
+        let mut builder = self.builder.try_lock_until(deadline).ok_or(rpc::TIMEOUT)?;
+        builder.create_vm(socket_path, timeout, deadline)
+    }
 }
 
 pub struct VmHandle {
-    inner: VmHandleInner,
+    inner: Mutex<VmHandleInner>,
+    timeout: Duration,
+    cancellation: watch::Sender<bool>,
 }
 
 struct VmHandleInner {
     runtime: Runtime,
     client: VmClient<tonic::transport::Channel>,
-    timeout: Duration,
     shares: BTreeMap<String, String>,
+    status: VmStatus,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+enum VmStatus {
+    Active,
+    RecoveryRequired,
 }
 
 impl VmConfigBuilder {
-    pub fn new() -> Self {
+    fn new() -> Self {
         Self {
-            inner: VmConfig::default(),
+            config: VmConfig::default(),
+            status: VmConfigStatus::Ready,
         }
     }
 
-    pub fn create_vm(&self, socket_path: String, timeout_ms: u32) -> Result<VmHandle, HRESULT> {
+    fn create_vm(
+        &mut self,
+        socket_path: String,
+        timeout: Duration,
+        deadline: Instant,
+    ) -> Result<VmHandle, HRESULT> {
+        if self.status == VmConfigStatus::CreationOutcomeUnknown {
+            return Err(rpc::INVALID_STATE);
+        }
         let runtime = match tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()
@@ -48,33 +103,36 @@ impl VmConfigBuilder {
             Ok(runtime) => runtime,
             Err(_) => return Err(E_FAIL),
         };
-        let timeout = Duration::from_millis(u64::from(timeout_ms));
-        let channel = match runtime.block_on(af_unix::connect_channel(socket_path.into(), timeout)) {
-            Ok(channel) => channel,
-            Err(error) if error.to_string().contains("timed out") => {
-                return Err(HRESULT::from_win32(WAIT_TIMEOUT.0));
-            }
-            Err(_) => return Err(HRESULT::from_win32(ERROR_CONNECTION_ABORTED.0)),
-        };
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(rpc::TIMEOUT);
+        }
+        let channel =
+            match runtime.block_on(af_unix::connect_channel(socket_path.into(), remaining)) {
+                Ok(channel) => channel,
+                Err(error) => return Err(rpc::io_error_to_hresult(&error)),
+            };
         let mut client = VmClient::new(channel);
         let request = CreateVmRequest {
-            config: Some(self.inner.clone()),
+            config: Some(self.config.clone()),
             log_id: String::new(),
         };
-        match runtime.block_on(async {
-            let mut request = Request::new(request);
-            request.set_timeout(timeout);
-            client.create_vm(request).await
-        }) {
-            Ok(_) => Ok(VmHandle {
-                inner: VmHandleInner {
-                    runtime,
-                    client,
-                    timeout,
-                    shares: BTreeMap::new(),
-                },
-            }),
-            Err(error) => Err(rpc_status_to_hresult(error)),
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(rpc::TIMEOUT);
+        }
+        match runtime.block_on(rpc::execute(
+            deadline,
+            None,
+            client.create_vm(request_with_timeout(request, remaining)),
+        )) {
+            Ok(_) => Ok(VmHandle::new(runtime, client, timeout)),
+            Err(error) => {
+                if error.uncertain {
+                    self.status = VmConfigStatus::CreationOutcomeUnknown;
+                }
+                Err(error.result)
+            }
         }
     }
 
@@ -94,12 +152,12 @@ impl VmConfigBuilder {
     }
 
     pub fn set_memory_mb(&mut self, memory_mb: u64) -> HRESULT {
-        self.inner.memory_config.get_or_insert_default().memory_mb = memory_mb;
+        self.config.memory_config.get_or_insert_default().memory_mb = memory_mb;
         S_OK
     }
 
     pub fn set_processor_count(&mut self, count: u32) -> HRESULT {
-        self.inner
+        self.config
             .processor_config
             .get_or_insert_default()
             .processor_count = count;
@@ -107,7 +165,7 @@ impl VmConfigBuilder {
     }
 
     pub fn set_hvsocket_path(&mut self, path: String) -> HRESULT {
-        self.inner.hvsocket_config.get_or_insert_default().path = path;
+        self.config.hvsocket_config.get_or_insert_default().path = path;
         S_OK
     }
 
@@ -118,7 +176,7 @@ impl VmConfigBuilder {
         host_path: String,
         read_only: bool,
     ) -> HRESULT {
-        self.inner
+        self.config
             .devices_config
             .get_or_insert_default()
             .scsi_disks
@@ -133,7 +191,7 @@ impl VmConfigBuilder {
     }
 
     pub fn set_consomme_nic(&mut self, nic_id: String, mac_address: String) -> HRESULT {
-        self.inner
+        self.config
             .devices_config
             .get_or_insert_default()
             .nic_config
@@ -149,18 +207,20 @@ impl VmConfigBuilder {
     }
 
     pub fn add_serial_port(&mut self, port: u32, pipe_name: String) -> HRESULT {
-        self.inner.serial_config.get_or_insert_default().ports.push(
-            vmservice::serial_config::Config {
+        self.config
+            .serial_config
+            .get_or_insert_default()
+            .ports
+            .push(vmservice::serial_config::Config {
                 port,
                 socket_path: pipe_name,
                 connect: true,
-            },
-        );
+            });
         S_OK
     }
 
     pub fn set_virtio_console_path(&mut self, path: String) -> HRESULT {
-        self.inner
+        self.config
             .devices_config
             .get_or_insert_default()
             .virtio_console = Some(VirtioConsoleConfig {
@@ -171,7 +231,7 @@ impl VmConfigBuilder {
     }
 
     fn direct_boot(&mut self) -> &mut DirectBoot {
-        let boot_config = self.inner.boot_config.get_or_insert_with(|| {
+        let boot_config = self.config.boot_config.get_or_insert_with(|| {
             vmservice::vm_config::BootConfig::DirectBoot(DirectBoot::default())
         });
         match boot_config {
@@ -182,24 +242,99 @@ impl VmConfigBuilder {
 }
 
 impl VmHandle {
-    pub fn resume_vm(&mut self) -> HRESULT {
-        self.empty_rpc(|client, request| Box::pin(async move { client.resume_vm(request).await }))
+    fn new(
+        runtime: Runtime,
+        client: VmClient<tonic::transport::Channel>,
+        timeout: Duration,
+    ) -> Self {
+        Self {
+            inner: Mutex::new(VmHandleInner {
+                runtime,
+                client,
+                shares: BTreeMap::new(),
+                status: VmStatus::Active,
+            }),
+            timeout,
+            cancellation: watch::channel(false).0,
+        }
     }
 
-    pub fn teardown_vm(&mut self) -> HRESULT {
-        let result = self.empty_rpc(|client, request| Box::pin(async move { client.teardown_vm(request).await }));
-        if result.is_ok() {
-            self.inner.shares.clear();
+    pub fn cancel_requests(&self) -> HRESULT {
+        self.cancellation.send_replace(true);
+        S_OK
+    }
+
+    fn with_operation(
+        &self,
+        cleanup: bool,
+        operation: impl FnOnce(&mut VmHandleInner, Instant, Option<watch::Receiver<bool>>) -> HRESULT,
+    ) -> HRESULT {
+        let deadline = Instant::now() + self.timeout;
+        let Some(mut inner) = self.inner.try_lock_until(deadline) else {
+            // No request was sent, so lock contention alone does not invalidate the VM.
+            return rpc::TIMEOUT;
+        };
+        if *self.cancellation.borrow() {
+            inner.status = VmStatus::RecoveryRequired;
+        }
+        if !cleanup && inner.status == VmStatus::RecoveryRequired {
+            return rpc::INVALID_STATE;
+        }
+        let cancellation = if cleanup {
+            None
+        } else {
+            Some(self.cancellation.subscribe())
+        };
+        let result = operation(&mut inner, deadline, cancellation);
+        // Cancellation can race a successful response, including during cleanup.
+        if *self.cancellation.borrow() {
+            inner.status = VmStatus::RecoveryRequired;
         }
         result
     }
 
-    pub fn quit(&mut self) -> HRESULT {
-        self.empty_rpc(|client, request| Box::pin(async move { client.quit(request).await }))
+    pub fn resume_vm(&self) -> HRESULT {
+        self.with_operation(false, |inner, deadline, cancellation| {
+            self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                (),
+                |mut client, request| async move { client.resume_vm(request).await },
+            )
+        })
+    }
+
+    pub fn teardown_vm(&self) -> HRESULT {
+        self.with_operation(true, |inner, deadline, cancellation| {
+            let result = self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                (),
+                |mut client, request| async move { client.teardown_vm(request).await },
+            );
+            if result.is_ok() {
+                inner.shares.clear();
+            }
+            result
+        })
+    }
+
+    pub fn quit(&self) -> HRESULT {
+        self.with_operation(true, |inner, deadline, cancellation| {
+            self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                (),
+                |mut client, request| async move { client.quit(request).await },
+            )
+        })
     }
 
     pub fn attach_scsi_disk(
-        &mut self,
+        &self,
         controller: u32,
         lun: u32,
         host_path: String,
@@ -208,155 +343,187 @@ impl VmHandle {
         self.modify_disk(ModifyType::Add, controller, lun, host_path, read_only)
     }
 
-    pub fn detach_scsi_disk(&mut self, controller: u32, lun: u32) -> HRESULT {
+    pub fn detach_scsi_disk(&self, controller: u32, lun: u32) -> HRESULT {
         self.modify_disk(ModifyType::Remove, controller, lun, String::new(), false)
     }
 
-    pub fn bind_port(&mut self, host_port: u16, guest_port: u16, tcp: bool, family: i32) -> HRESULT {
+    pub fn bind_port(&self, host_port: u16, guest_port: u16, tcp: bool, family: i32) -> HRESULT {
         self.modify_port(ModifyType::Update, host_port, guest_port, tcp, family)
     }
 
-    pub fn unbind_port(&mut self, host_port: u16, guest_port: u16, tcp: bool, family: i32) -> HRESULT {
+    pub fn unbind_port(&self, host_port: u16, guest_port: u16, tcp: bool, family: i32) -> HRESULT {
         self.modify_port(ModifyType::Remove, host_port, guest_port, tcp, family)
     }
 
-    pub fn add_share(&mut self, tag: String, host_path: String, read_only: bool) -> HRESULT {
-        if self.inner.shares.contains_key(&tag) {
-            return HRESULT::from_win32(ERROR_ALREADY_EXISTS.0);
-        }
-        let instance_id = match GUID::new() {
-            Ok(instance_id) => format!("{instance_id:?}"),
-            Err(error) => return error.code(),
-        };
-        let request = vmservice::AddVpciDeviceRequest {
-            instance_id: instance_id.clone(),
-            device: Some(vmservice::PcieDeviceKind {
-                kind: Some(vmservice::pcie_device_kind::Kind::Virtio(vmservice::VirtioDevice {
-                    kind: Some(vmservice::virtio_device::Kind::Fs(vmservice::VirtioFs {
-                        tag: tag.clone(),
-                        root_path: host_path,
-                        read_only,
-                    })),
-                })),
-            }),
-        };
-        match self.inner.runtime.block_on(self.inner.client.add_vpci_device(
-            request_with_timeout(request, self.inner.timeout),
-        )) {
-            Ok(_) => {
-                self.inner.shares.insert(tag, instance_id);
-                S_OK
+    pub fn add_share(&self, tag: String, host_path: String, read_only: bool) -> HRESULT {
+        self.with_operation(false, |inner, deadline, cancellation| {
+            if inner.shares.contains_key(&tag) {
+                return HRESULT::from_win32(ERROR_ALREADY_EXISTS.0);
             }
-            Err(error) => rpc_status_to_hresult(error),
-        }
+            let instance_id = match GUID::new() {
+                Ok(instance_id) => format!("{instance_id:?}"),
+                Err(error) => return error.code(),
+            };
+            let request = vmservice::AddVpciDeviceRequest {
+                instance_id: instance_id.clone(),
+                device: Some(vmservice::PcieDeviceKind {
+                    kind: Some(vmservice::pcie_device_kind::Kind::Virtio(
+                        vmservice::VirtioDevice {
+                            kind: Some(vmservice::virtio_device::Kind::Fs(vmservice::VirtioFs {
+                                tag: tag.clone(),
+                                root_path: host_path,
+                                read_only,
+                            })),
+                        },
+                    )),
+                }),
+            };
+            let result = self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                request,
+                |mut client, request| async move { client.add_vpci_device(request).await },
+            );
+            if result.is_ok() {
+                inner.shares.insert(tag, instance_id);
+            }
+            result
+        })
     }
 
-    pub fn remove_share(&mut self, tag: &str) -> HRESULT {
-        let Some(instance_id) = self.inner.shares.get(tag) else {
-            return HRESULT::from_win32(ERROR_NOT_FOUND.0);
-        };
-        let request = vmservice::RemoveVpciDeviceRequest {
-            instance_id: instance_id.clone(),
-        };
-        match self.inner.runtime.block_on(self.inner.client.remove_vpci_device(
-            request_with_timeout(request, self.inner.timeout),
-        )) {
-            Ok(_) => {
-                self.inner.shares.remove(tag);
-                S_OK
+    pub fn remove_share(&self, tag: &str) -> HRESULT {
+        self.with_operation(false, |inner, deadline, cancellation| {
+            let Some(instance_id) = inner.shares.get(tag) else {
+                return HRESULT::from_win32(ERROR_NOT_FOUND.0);
+            };
+            let request = vmservice::RemoveVpciDeviceRequest {
+                instance_id: instance_id.clone(),
+            };
+            let result = self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                request,
+                |mut client, request| async move { client.remove_vpci_device(request).await },
+            );
+            if result.is_ok() {
+                inner.shares.remove(tag);
             }
-            Err(error) => rpc_status_to_hresult(error),
-        }
+            result
+        })
     }
 
-    fn empty_rpc<F>(&mut self, operation: F) -> HRESULT
+    fn rpc<T, F, R>(
+        &self,
+        inner: &mut VmHandleInner,
+        deadline: Instant,
+        cancellation: Option<watch::Receiver<bool>>,
+        message: T,
+        operation: F,
+    ) -> HRESULT
     where
-        F: for<'a> FnOnce(
-            &'a mut VmClient<tonic::transport::Channel>,
-            Request<()>,
-        ) -> std::pin::Pin<
-            Box<dyn std::future::Future<Output = Result<tonic::Response<()>, tonic::Status>> + 'a>,
-        >,
+        F: FnOnce(VmClient<tonic::transport::Channel>, Request<T>) -> R,
+        R: Future<Output = Result<tonic::Response<()>, tonic::Status>>,
     {
-        match self.inner.runtime.block_on(operation(
-            &mut self.inner.client,
-            request_with_timeout((), self.inner.timeout),
-        )) {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return rpc::TIMEOUT;
+        }
+        let result = inner.runtime.block_on(rpc::execute(
+            deadline,
+            cancellation,
+            operation(
+                inner.client.clone(),
+                request_with_timeout(message, remaining),
+            ),
+        ));
+        match result {
             Ok(_) => S_OK,
-            Err(error) => rpc_status_to_hresult(error),
+            Err(error) => {
+                if error.uncertain {
+                    inner.status = VmStatus::RecoveryRequired;
+                }
+                error.result
+            }
         }
     }
 
     fn modify_disk(
-        &mut self,
+        &self,
         modify_type: ModifyType,
         controller: u32,
         lun: u32,
         host_path: String,
         read_only: bool,
     ) -> HRESULT {
-        let request = ModifyResourceRequest {
-            r#type: modify_type as i32,
-            resource: Some(vmservice::modify_resource_request::Resource::ScsiDisk(
-                ScsiDisk {
-                    controller,
-                    lun,
-                    r#type: disk_type(&host_path),
-                    host_path,
-                    read_only,
-                },
-            )),
-        };
-        self.modify_resource(request)
+        self.with_operation(false, |inner, deadline, cancellation| {
+            let request = ModifyResourceRequest {
+                r#type: modify_type as i32,
+                resource: Some(vmservice::modify_resource_request::Resource::ScsiDisk(
+                    ScsiDisk {
+                        controller,
+                        lun,
+                        r#type: disk_type(&host_path),
+                        host_path,
+                        read_only,
+                    },
+                )),
+            };
+            self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                request,
+                |mut client, request| async move { client.modify_resource(request).await },
+            )
+        })
     }
 
     fn modify_port(
-        &mut self,
+        &self,
         modify_type: ModifyType,
         host_port: u16,
         guest_port: u16,
         tcp: bool,
         family: i32,
     ) -> HRESULT {
-        let host_address = match family {
-            family if family == i32::from(AF_INET) => "127.0.0.1",
-            family if family == i32::from(AF_INET6) => "::1",
-            _ => return E_INVALIDARG,
-        };
-        let protocol = if tcp {
-            vmservice::IpProtocol::Tcp
-        } else {
-            vmservice::IpProtocol::Udp
-        };
-        let request = ModifyResourceRequest {
-            r#type: modify_type as i32,
-            resource: Some(vmservice::modify_resource_request::Resource::NicConfig(
-                NicConfig {
-                    backend: Some(vmservice::nic_config::Backend::Consomme(ConsommeBackend {
-                        cidr: String::new(),
-                        ports: vec![PortConfig {
-                            host_port: u32::from(host_port),
-                            guest_port: u32::from(guest_port),
-                            protocol: protocol as i32,
-                            host_address: host_address.to_string(),
-                        }],
-                    })),
-                    ..Default::default()
-                },
-            )),
-        };
-        self.modify_resource(request)
-    }
-
-    fn modify_resource(&mut self, request: ModifyResourceRequest) -> HRESULT {
-        match self.inner.runtime.block_on(
-            self.inner
-                .client
-                .modify_resource(request_with_timeout(request, self.inner.timeout)),
-        ) {
-            Ok(_) => S_OK,
-            Err(error) => rpc_status_to_hresult(error),
-        }
+        self.with_operation(false, |inner, deadline, cancellation| {
+            let host_address = match family {
+                family if family == i32::from(AF_INET) => "127.0.0.1",
+                family if family == i32::from(AF_INET6) => "::1",
+                _ => return E_INVALIDARG,
+            };
+            let protocol = if tcp {
+                vmservice::IpProtocol::Tcp
+            } else {
+                vmservice::IpProtocol::Udp
+            };
+            let request = ModifyResourceRequest {
+                r#type: modify_type as i32,
+                resource: Some(vmservice::modify_resource_request::Resource::NicConfig(
+                    NicConfig {
+                        backend: Some(vmservice::nic_config::Backend::Consomme(ConsommeBackend {
+                            cidr: String::new(),
+                            ports: vec![PortConfig {
+                                host_port: u32::from(host_port),
+                                guest_port: u32::from(guest_port),
+                                protocol: protocol as i32,
+                                host_address: host_address.to_string(),
+                            }],
+                        })),
+                        ..Default::default()
+                    },
+                )),
+            };
+            self.rpc(
+                inner,
+                deadline,
+                cancellation,
+                request,
+                |mut client, request| async move { client.modify_resource(request).await },
+            )
+        })
     }
 }
 
@@ -373,11 +540,3 @@ fn disk_type(path: &str) -> i32 {
         DiskType::ScsiDiskTypeVhd1 as i32
     }
 }
-
-fn rpc_status_to_hresult(error: tonic::Status) -> HRESULT {
-    if error.code() == tonic::Code::DeadlineExceeded {
-        HRESULT::from_win32(WAIT_TIMEOUT.0)
-    } else {
-        HRESULT::from_win32(ERROR_CONNECTION_ABORTED.0)
-    }
-}

src/windows/wslopenvmm/src/lib.rs

@@ -2,20 +2,26 @@
 
 #![expect(clippy::missing_safety_doc)]
 
+mod af_unix;
 mod client;
-pub mod af_unix;
+mod rpc;
 
-pub mod vmservice {
+mod vmservice {
     tonic::include_proto!("vmservice");
 }
 
-use client::{VmConfigBuilder, VmHandle};
-use parking_lot::Mutex;
+use client::{VmConfigBuilder, VmConfigHandle, VmHandle};
 use windows::Win32::Foundation::{E_INVALIDARG, E_POINTER, S_OK};
 use windows::core::{HRESULT, PCWSTR};
 
-pub struct WslOpenVmmConfig(Mutex<VmConfigBuilder>);
-pub struct WslOpenVmmVm(Mutex<VmHandle>);
+pub struct WslOpenVmmConfig(VmConfigHandle);
+pub struct WslOpenVmmVm(VmHandle);
+
+impl WslOpenVmmVm {
+    fn new(client: VmHandle) -> Self {
+        Self(client)
+    }
+}
 
 unsafe fn string_from_wide(value: *const u16) -> Result<String, HRESULT> {
     if value.is_null() {
@@ -32,17 +38,17 @@ unsafe fn with_config(
     let Some(config) = (unsafe { handle.as_ref() }) else {
         return E_POINTER;
     };
-    operation(&mut config.0.lock())
+    config.0.update(operation)
 }
 
 unsafe fn with_vm(
     handle: *mut WslOpenVmmVm,
-    operation: impl FnOnce(&mut VmHandle) -> HRESULT,
+    operation: impl FnOnce(&VmHandle) -> HRESULT,
 ) -> HRESULT {
     let Some(vm) = (unsafe { handle.as_ref() }) else {
         return E_POINTER;
     };
-    operation(&mut vm.0.lock())
+    operation(&vm.0)
 }
 
 #[unsafe(no_mangle)]
@@ -52,9 +58,7 @@ pub unsafe extern "C" fn WslOpenVmmCreateConfig(config: *mut *mut WslOpenVmmConf
     }
 
     unsafe {
-        *config = Box::into_raw(Box::new(WslOpenVmmConfig(Mutex::new(
-            VmConfigBuilder::new(),
-        ))));
+        *config = Box::into_raw(Box::new(WslOpenVmmConfig(VmConfigHandle::new())));
     }
     S_OK.0
 }
@@ -220,21 +224,17 @@ pub unsafe extern "C" fn WslOpenVmmCreateVm(
         Err(error) => return error.0,
     };
 
-    let config_box = unsafe { Box::from_raw(config_handle) };
-    let result = config_box.0.lock().create_vm(socket_path, timeout_ms);
+    let config_ref = unsafe { &*config_handle };
+    let result = config_ref.0.create_vm(socket_path, timeout_ms);
     let client = match result {
         Ok(client) => client,
-        Err(error) => {
-            unsafe {
-                *config = Box::into_raw(config_box);
-            }
-            return error.0;
-        }
+        Err(error) => return error.0,
     };
 
     unsafe {
         *config = std::ptr::null_mut();
-        *vm = Box::into_raw(Box::new(WslOpenVmmVm(Mutex::new(client))));
+        drop(Box::from_raw(config_handle));
+        *vm = Box::into_raw(Box::new(WslOpenVmmVm::new(client)));
     }
     S_OK.0
 }
@@ -246,6 +246,11 @@ pub unsafe extern "C" fn WslOpenVmmDestroyVm(vm: *mut WslOpenVmmVm) {
     }
 }
 
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn WslOpenVmmVmCancelRequests(vm: *mut WslOpenVmmVm) -> i32 {
+    unsafe { with_vm(vm, VmHandle::cancel_requests).0 }
+}
+
 #[unsafe(no_mangle)]
 pub unsafe extern "C" fn WslOpenVmmVmResume(vm: *mut WslOpenVmmVm) -> i32 {
     unsafe { with_vm(vm, VmHandle::resume_vm).0 }
@@ -298,7 +303,12 @@ pub unsafe extern "C" fn WslOpenVmmVmBindPort(
     tcp: i32,
     family: i32,
 ) -> i32 {
-    unsafe { with_vm(vm, |vm| vm.bind_port(host_port, guest_port, tcp != 0, family)).0 }
+    unsafe {
+        with_vm(vm, |vm| {
+            vm.bind_port(host_port, guest_port, tcp != 0, family)
+        })
+        .0
+    }
 }
 
 #[unsafe(no_mangle)]
@@ -309,7 +319,12 @@ pub unsafe extern "C" fn WslOpenVmmVmUnbindPort(
     tcp: i32,
     family: i32,
 ) -> i32 {
-    unsafe { with_vm(vm, |vm| vm.unbind_port(host_port, guest_port, tcp != 0, family)).0 }
+    unsafe {
+        with_vm(vm, |vm| {
+            vm.unbind_port(host_port, guest_port, tcp != 0, family)
+        })
+        .0
+    }
 }
 
 #[unsafe(no_mangle)]
@@ -331,10 +346,7 @@ pub unsafe extern "C" fn WslOpenVmmVmAddShare(
 }
 
 #[unsafe(no_mangle)]
-pub unsafe extern "C" fn WslOpenVmmVmRemoveShare(
-    vm: *mut WslOpenVmmVm,
-    tag: *const u16,
-) -> i32 {
+pub unsafe extern "C" fn WslOpenVmmVmRemoveShare(vm: *mut WslOpenVmmVm, tag: *const u16) -> i32 {
     let tag = match unsafe { string_from_wide(tag) } {
         Ok(tag) => tag,
         Err(error) => return error.0,

src/windows/wslopenvmm/src/rpc.rs

@@ -0,0 +1,212 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+
+use std::error::Error;
+use std::future::Future;
+use std::io;
+use std::time::Instant;
+
+use tokio::sync::watch;
+use tonic::Code;
+use windows::Win32::Foundation::{
+    E_ABORT, E_ACCESSDENIED, E_FAIL, E_INVALIDARG, E_NOTIMPL, ERROR_ALREADY_EXISTS,
+    ERROR_CONNECTION_ABORTED, ERROR_INVALID_DATA, ERROR_INVALID_STATE, ERROR_LOGON_FAILURE,
+    ERROR_NOT_ENOUGH_MEMORY, ERROR_NOT_FOUND, ERROR_RETRY, WAIT_TIMEOUT,
+};
+use windows::core::HRESULT;
+
+pub const INVALID_STATE: HRESULT = HRESULT::from_win32(ERROR_INVALID_STATE.0);
+pub const TIMEOUT: HRESULT = HRESULT::from_win32(WAIT_TIMEOUT.0);
+
+pub struct RpcFailure {
+    pub result: HRESULT,
+    pub uncertain: bool,
+}
+
+pub async fn execute<T>(
+    deadline: Instant,
+    cancellation: Option<watch::Receiver<bool>>,
+    operation: impl Future<Output = Result<T, tonic::Status>>,
+) -> Result<T, RpcFailure> {
+    // A closed sender also cancels the operation; no lifecycle owner is required here.
+    let cancellation = async {
+        if let Some(mut cancellation) = cancellation {
+            let _ = cancellation.wait_for(|cancelled| *cancelled).await;
+        } else {
+            std::future::pending::<()>().await;
+        }
+    };
+    // Prefer the local deadline when the peer's cancellation arrives in the same poll.
+    // This also bounds peers that ignore grpc-timeout.
+    tokio::select! {
+        biased;
+        _ = tokio::time::sleep_until(deadline.into()) =>
+            Err(RpcFailure { result: TIMEOUT, uncertain: true }),
+        _ = cancellation =>
+            Err(RpcFailure { result: E_ABORT, uncertain: true }),
+        result = operation => result.map_err(failure_from_status),
+    }
+}
+
+fn failure_from_status(status: tonic::Status) -> RpcFailure {
+    let mut source = status.source();
+    while let Some(error) = source {
+        // Tonic reports its local request timeout with Code::Cancelled.
+        if error.is::<tonic::TimeoutExpired>() {
+            return RpcFailure {
+                result: TIMEOUT,
+                uncertain: true,
+            };
+        }
+        source = error.source();
+    }
+    RpcFailure {
+        result: status_to_hresult(status.code()),
+        uncertain: !is_definite_rejection(status.code()),
+    }
+}
+
+fn status_to_hresult(code: Code) -> HRESULT {
+    match code {
+        Code::Ok => windows::Win32::Foundation::S_OK,
+        Code::Cancelled => E_ABORT,
+        Code::InvalidArgument | Code::OutOfRange => E_INVALIDARG,
+        Code::DeadlineExceeded => TIMEOUT,
+        Code::NotFound => HRESULT::from_win32(ERROR_NOT_FOUND.0),
+        Code::AlreadyExists => HRESULT::from_win32(ERROR_ALREADY_EXISTS.0),
+        Code::PermissionDenied => E_ACCESSDENIED,
+        Code::Unauthenticated => HRESULT::from_win32(ERROR_LOGON_FAILURE.0),
+        Code::ResourceExhausted => HRESULT::from_win32(ERROR_NOT_ENOUGH_MEMORY.0),
+        Code::FailedPrecondition => INVALID_STATE,
+        Code::Aborted => HRESULT::from_win32(ERROR_RETRY.0),
+        Code::Unimplemented => E_NOTIMPL,
+        Code::Unavailable => HRESULT::from_win32(ERROR_CONNECTION_ABORTED.0),
+        Code::DataLoss => HRESULT::from_win32(ERROR_INVALID_DATA.0),
+        Code::Unknown | Code::Internal => E_FAIL,
+    }
+}
+
+fn is_definite_rejection(code: Code) -> bool {
+    // Only validation/authorization rejections are safe to correct and resubmit.
+    // Internal, aborted, resource exhaustion and transport errors may follow a mutation.
+    matches!(
+        code,
+        Code::InvalidArgument
+            | Code::OutOfRange
+            | Code::NotFound
+            | Code::AlreadyExists
+            | Code::PermissionDenied
+            | Code::Unauthenticated
+            | Code::FailedPrecondition
+            | Code::Unimplemented
+    )
+}
+
+pub fn io_error_to_hresult(error: &io::Error) -> HRESULT {
+    match error.kind() {
+        io::ErrorKind::TimedOut => TIMEOUT,
+        io::ErrorKind::PermissionDenied => E_ACCESSDENIED,
+        io::ErrorKind::InvalidInput => E_INVALIDARG,
+        _ => match error.raw_os_error() {
+            Some(code) => HRESULT::from_win32(code as u32),
+            None => HRESULT::from_win32(ERROR_CONNECTION_ABORTED.0),
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn execution_accepts_cancellation_without_vm_state() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        for cancelled in [true, false] {
+            let (send, receive) = watch::channel(cancelled);
+            if !cancelled {
+                drop(send);
+            }
+            let result = runtime.block_on(execute::<()>(
+                Instant::now() + std::time::Duration::from_secs(1),
+                Some(receive),
+                std::future::pending(),
+            ));
+            let error = result.unwrap_err();
+            assert_eq!(error.result, E_ABORT);
+            assert!(error.uncertain);
+        }
+    }
+
+    #[test]
+    fn local_deadline_does_not_depend_on_peer_timeout_support() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let start = Instant::now();
+        let result = runtime.block_on(execute::<()>(
+            start + std::time::Duration::from_millis(50),
+            None,
+            std::future::pending(),
+        ));
+        let error = result.unwrap_err();
+        assert_eq!(error.result, TIMEOUT);
+        assert!(error.uncertain);
+        assert!(start.elapsed() < std::time::Duration::from_secs(2));
+    }
+
+    #[test]
+    fn status_and_io_errors_have_explicit_hresult_categories() {
+        for (code, expected) in [
+            (Code::Ok, windows::Win32::Foundation::S_OK),
+            (Code::Cancelled, E_ABORT),
+            (Code::Unknown, E_FAIL),
+            (Code::InvalidArgument, E_INVALIDARG),
+            (Code::DeadlineExceeded, TIMEOUT),
+            (Code::NotFound, HRESULT::from_win32(ERROR_NOT_FOUND.0)),
+            (
+                Code::AlreadyExists,
+                HRESULT::from_win32(ERROR_ALREADY_EXISTS.0),
+            ),
+            (Code::PermissionDenied, E_ACCESSDENIED),
+            (
+                Code::ResourceExhausted,
+                HRESULT::from_win32(ERROR_NOT_ENOUGH_MEMORY.0),
+            ),
+            (Code::FailedPrecondition, INVALID_STATE),
+            (Code::Aborted, HRESULT::from_win32(ERROR_RETRY.0)),
+            (Code::OutOfRange, E_INVALIDARG),
+            (Code::Unimplemented, E_NOTIMPL),
+            (Code::Internal, E_FAIL),
+            (
+                Code::Unavailable,
+                HRESULT::from_win32(ERROR_CONNECTION_ABORTED.0),
+            ),
+            (Code::DataLoss, HRESULT::from_win32(ERROR_INVALID_DATA.0)),
+            (
+                Code::Unauthenticated,
+                HRESULT::from_win32(ERROR_LOGON_FAILURE.0),
+            ),
+        ] {
+            assert_eq!(status_to_hresult(code), expected);
+        }
+        assert_eq!(
+            io_error_to_hresult(&io::Error::from(io::ErrorKind::TimedOut)),
+            TIMEOUT
+        );
+        assert_eq!(
+            io_error_to_hresult(&io::Error::from(io::ErrorKind::InvalidInput)),
+            E_INVALIDARG
+        );
+        assert_eq!(
+            io_error_to_hresult(&io::Error::from(io::ErrorKind::PermissionDenied)),
+            E_ACCESSDENIED
+        );
+        assert_eq!(
+            io_error_to_hresult(&io::Error::from_raw_os_error(10054)),
+            HRESULT::from_win32(10054)
+        );
+    }
+}

src/windows/wslopenvmm/wslopenvmm.h

@@ -13,6 +13,7 @@ typedef struct WslOpenVmmConfig WslOpenVmmConfig;
 typedef struct WslOpenVmmVm WslOpenVmmVm;
 
 // Config and VM handles are opaque Rust-owned objects. Destroy each returned handle exactly once.
+// Destruction (and successful CreateVm consumption of Config) must not race any use of that handle.
 __declspec(dllimport) HRESULT WslOpenVmmCreateConfig(_Out_ WslOpenVmmConfig** Config);
 __declspec(dllimport) void WslOpenVmmDestroyConfig(_In_opt_ WslOpenVmmConfig* Config);
 __declspec(dllimport) HRESULT WslOpenVmmConfigSetKernelPath(_Inout_ WslOpenVmmConfig* Config, _In_ LPCWSTR Path);
@@ -26,9 +27,15 @@ __declspec(dllimport) HRESULT WslOpenVmmConfigSetConsommeNic(_Inout_ WslOpenVmmC
 __declspec(dllimport) HRESULT WslOpenVmmConfigAddSerialPort(_Inout_ WslOpenVmmConfig* Config, _In_ UINT32 Port, _In_ LPCWSTR PipeName);
 __declspec(dllimport) HRESULT WslOpenVmmConfigSetVirtioConsolePath(_Inout_ WslOpenVmmConfig* Config, _In_ LPCWSTR Path);
 // On success, this consumes Config and sets it to nullptr. On failure, Config remains valid and Vm is nullptr.
+// TimeoutMs must be nonzero and bounds lock acquisition, connection establishment, and CreateVm together.
+// After an ambiguous CreateVm failure, discard both the config and the owning process before retrying.
 __declspec(dllimport) HRESULT WslOpenVmmCreateVm(_Inout_ WslOpenVmmConfig** Config, _In_ LPCWSTR SocketPath, _In_ UINT32 TimeoutMs, _Out_ WslOpenVmmVm** Vm);
 
 __declspec(dllimport) void WslOpenVmmDestroyVm(_In_opt_ WslOpenVmmVm* Vm);
+// VM calls use TimeoutMs from creation, including time waiting for the per-VM lock.
+// Ambiguous RPC failures permanently invalidate mutations; teardown/quit remain available with deadlines.
+// Cancellation is nonblocking and may race an RPC, but not destruction. It also requires VM recreation.
+__declspec(dllimport) HRESULT WslOpenVmmVmCancelRequests(_Inout_ WslOpenVmmVm* Vm);
 __declspec(dllimport) HRESULT WslOpenVmmVmResume(_Inout_ WslOpenVmmVm* Vm);
 __declspec(dllimport) HRESULT WslOpenVmmVmTeardown(_Inout_ WslOpenVmmVm* Vm);
 __declspec(dllimport) HRESULT WslOpenVmmVmQuit(_Inout_ WslOpenVmmVm* Vm);