feat: Add sqlx runtime implementations for Wasip3 trait and minor runtime adjustments

Signed-off-by: Aditya <aditya.salunkh919@gmail.com>

Author: Aditya1404Sal

sqlx-core/src/fs.rs

@@ -50,7 +50,7 @@ pub async fn create_dir_all<P: AsRef<Path>>(path: P) -> io::Result<()> {
     let path = PathBuf::from(path.as_ref());
     #[cfg(not(target_arch = "wasm32"))]
     {
-    rt::spawn_blocking(move || std::fs::create_dir_all(path)).await
+        rt::spawn_blocking(move || std::fs::create_dir_all(path)).await
     }
     #[cfg(target_arch = "wasm32")]
     {
@@ -103,7 +103,6 @@ pub async fn read_dir(path: PathBuf) -> io::Result<ReadDir> {
     })
 }
 
-
 #[cfg(target_arch = "wasm32")]
 pub async fn read_dir(path: PathBuf) -> io::Result<ReadDir> {
     todo!()
@@ -144,8 +143,6 @@ impl ReadDir {
             } else {
                 Ok(None)
             }
-
-
         }
         #[cfg(target_arch = "wasm32")]
         todo!()

sqlx-core/src/migrate/mod.rs

@@ -17,6 +17,5 @@ pub use source::{MigrationSource, ResolveConfig, ResolveWith};
 #[doc(hidden)]
 pub use source::resolve;
 #[cfg(not(target_arch = "wasm32"))]
-
 #[doc(hidden)]
 pub use source::{resolve_blocking, resolve_blocking_with_config};

sqlx-core/src/net/socket/mod.rs

@@ -186,7 +186,7 @@ pub async fn connect_tcp<Ws: WithSocket>(
     port: u16,
     with_socket: Ws,
 ) -> crate::Result<Ws::Output> {
-    #[cfg(all(feature = "_rt-tokio", not(target_arch = "wasm32")))]    
+    #[cfg(all(feature = "_rt-tokio", not(target_arch = "wasm32")))]
     if crate::rt::rt_tokio::available() {
         return Ok(with_socket
             .with_socket(tokio::net::TcpStream::connect((host, port)).await?)
@@ -206,7 +206,6 @@ pub async fn connect_tcp<Ws: WithSocket>(
         let res = crate::rt::rt_wasip3::connect_tcp(host, port, with_socket).await;
         return res;
     }
-
 }
 
 /// Open a TCP socket to `host` and `port`.

sqlx-core/src/rt/mod.rs

@@ -26,6 +26,10 @@ pub enum JoinHandle<T> {
     #[cfg(feature = "_rt-tokio")]
     Tokio(tokio::task::JoinHandle<T>),
 
+    // WASI P3 runtime
+    #[cfg(target_arch = "wasm32")]
+    Wasip3(crate::rt::rt_wasip3::JoinHandle<T>),
+
     // Implementation shared by `smol` and `async-global-executor`
     #[cfg(feature = "_rt-async-task")]
     AsyncTask(Option<async_task::Task<T>>),
@@ -37,30 +41,20 @@ pub enum JoinHandle<T> {
 pub async fn timeout<F: Future>(duration: Duration, f: F) -> Result<F::Output, TimeoutError> {
     #[cfg(target_arch = "wasm32")]
     {
-        let timeout = crate::rt::rt_wasip3::spawn(wasi::clocks::monotonic_clock::wait_for(
+        let timeout_future = wasip3::clocks::monotonic_clock::wait_for(
             duration.as_nanos().try_into().unwrap_or(u64::MAX),
-        ));
-        let mut timeout = core::pin::pin!(timeout);
+        );
+        let mut timeout = core::pin::pin!(timeout_future);
         let mut f = core::pin::pin!(f);
-        core::future::poll_fn(|cx| {
-            match timeout.as_mut().poll(cx) {
-                Poll::Ready(Some(())) => {
-                    Poll::Ready(Err(TimeoutError(())))
-                }
-                Poll::Ready(None) => {
-                    Poll::Ready(Err(TimeoutError(())))
-                }
-                Poll::Pending => {
-                    f.as_mut().poll(cx).map(Ok)
-                }
-            }
+
+        return core::future::poll_fn(|cx| match timeout.as_mut().poll(cx) {
+            Poll::Ready(_) => Poll::Ready(Err(TimeoutError)),
+            Poll::Pending => f.as_mut().poll(cx).map(Ok),
         })
-        .await
+        .await;
     }
 
-    #[cfg(all(feature = "_rt-tokio", not(target_arch = "wasm32")))]
-    #[cfg(debug_assertions)]
-    let f = Box::pin(f);
+    #[cfg(feature = "_rt-tokio")]
     if rt_tokio::available() {
         return tokio::time::timeout(duration, f)
             .await
@@ -80,11 +74,11 @@ pub async fn timeout<F: Future>(duration: Duration, f: F) -> Result<F::Output, T
 pub async fn sleep(duration: Duration) {
     #[cfg(target_arch = "wasm32")]
     {
-        return crate::rt::rt_wasip3::spawn(wasi::clocks::monotonic_clock::wait_for(
+        wasip3::clocks::monotonic_clock::wait_for(
             duration.as_nanos().try_into().unwrap_or(u64::MAX),
-        ))
-        .await
-        .unwrap();
+        )
+        .await;
+        return;
     }
     #[cfg(feature = "_rt-tokio")]
     if rt_tokio::available() {
@@ -107,6 +101,11 @@ where
     F: Future + Send + 'static,
     F::Output: Send + 'static,
 {
+    #[cfg(target_arch = "wasm32")]
+    {
+        return JoinHandle::Wasip3(crate::rt::rt_wasip3::spawn(fut));
+    }
+
     #[cfg(feature = "_rt-tokio")]
     if let Ok(handle) = tokio::runtime::Handle::try_current() {
         return JoinHandle::Tokio(handle.spawn(fut));
@@ -124,15 +123,6 @@ where
         }
     }
 }
-#[cfg(target_arch = "wasm32")]
-#[track_caller]
-pub fn spawn<F>(fut: F) -> JoinHandle<F::Output>
-where
-    F: Future + Send + 'static,
-    F::Output: Send + 'static,
-{
-    JoinHandle::Tokio(tokio::task::spawn(fut))
-}
 
 #[cfg(target_arch = "wasm32")]
 #[track_caller]
@@ -141,8 +131,9 @@ where
     F: FnOnce() -> R + Send + 'static,
     R: Send + 'static,
 {
-    // In WASI P3, we use our async spawn_blocking implementation
-    JoinHandle::Tokio(tokio::task::spawn(crate::rt::rt_wasip3::spawn_blocking(f)))
+    JoinHandle::Wasip3(crate::rt::rt_wasip3::spawn(
+        crate::rt::rt_wasip3::spawn_blocking(f),
+    ))
 }
 
 #[cfg(not(target_arch = "wasm32"))]
@@ -175,7 +166,7 @@ pub async fn yield_now() {
     {
         return crate::rt::rt_wasip3::yield_now().await;
     }
-    
+
     #[cfg(feature = "_rt-tokio")]
     if rt_tokio::available() {
         return tokio::task::yield_now().await;
@@ -204,14 +195,15 @@ pub async fn yield_now() {
     .await
 }
 
+#[cfg(not(target_arch = "wasm32"))]
 #[track_caller]
 pub fn test_block_on<F: Future>(f: F) -> F::Output {
     #[cfg(feature = "_rt-async-io")]
     {
         return async_io::block_on(f);
     }
 
-    #[cfg(any(feature = "_rt-tokio", target_arch = "wasm32"))]
+    #[cfg(feature = "_rt-tokio")]
     {
         return tokio::runtime::Builder::new_current_thread()
             .enable_all()
@@ -223,7 +215,7 @@ pub fn test_block_on<F: Future>(f: F) -> F::Output {
     #[cfg(all(
         feature = "_rt-async-std",
         not(feature = "_rt-async-io"),
-        not(any(feature = "_rt-tokio", target_arch = "wasm32"))
+        not(feature = "_rt-tokio")
     ))]
     {
         return async_std::task::block_on(f);
@@ -232,14 +224,22 @@ pub fn test_block_on<F: Future>(f: F) -> F::Output {
     #[cfg(not(any(
         feature = "_rt-async-io",
         feature = "_rt-async-std",
-        feature = "_rt-tokio",
-        target_arch = "wasm32"
+        feature = "_rt-tokio"
     )))]
     {
         missing_rt(f)
     }
 }
 
+#[cfg(target_arch = "wasm32")]
+#[track_caller]
+pub fn test_block_on<F: Future + 'static>(f: F) -> F::Output
+where
+    F::Output: 'static,
+{
+    wasip3::wit_bindgen::rt::async_support::block_on(f)
+}
+
 #[track_caller]
 pub const fn missing_rt<T>(_unused: T) -> ! {
     if cfg!(feature = "_rt-tokio") {
@@ -264,11 +264,14 @@ impl<T: Send + 'static> Future for JoinHandle<T> {
                 .expect("BUG: task taken")
                 .poll(cx),
 
-            #[cfg(any(feature = "_rt-tokio",target_arch = "wasm32"))]
+            #[cfg(feature = "_rt-tokio")]
             Self::Tokio(handle) => Pin::new(handle)
                 .poll(cx)
                 .map(|res| res.expect("spawned task panicked")),
 
+            #[cfg(target_arch = "wasm32")]
+            Self::Wasip3(handle) => Pin::new(handle).poll(cx),
+
             Self::_Phantom(_) => {
                 let _ = cx;
                 unreachable!("runtime should have been checked on spawn")

sqlx-core/src/rt/rt_wasip3/mod.rs

@@ -2,12 +2,11 @@ use core::future::Future;
 use core::pin::Pin;
 use core::task::{Context, Poll};
 
+use crate::net::WithSocket;
 use bytes::{Buf, BytesMut};
 use wasip3::sockets::types::{IpAddressFamily, IpSocketAddress, TcpSocket as WasiTcpSocket};
-use wasip3::wit_stream;
 use wasip3::wit_bindgen::StreamResult;
-
-use crate::net::WithSocket;
+use wasip3::wit_stream;
 
 mod socket;
 
@@ -28,14 +27,12 @@ impl<T> Future for JoinHandle<T> {
 pub fn spawn<T: 'static + Send>(fut: impl Future<Output = T> + Send + 'static) -> JoinHandle<T> {
     JoinHandle {
         future: Box::pin(async move {
-            // Yield to allow other tasks to run cooperatively
-            wasip3::wit_bindgen::yield_async().await;            
+            wasip3::wit_bindgen::yield_async().await;
             fut.await
         }),
     }
 }
 
-// CPU-intensive operations using wit_bindgen's yield_blocking
 pub fn spawn_blocking<F, R>(f: F) -> impl Future<Output = R>
 where
     F: FnOnce() -> R + Send + 'static,
@@ -48,22 +45,34 @@ where
     }
 }
 
-// Native async yielding
+// Cooperative yielding for WASI P3
 pub async fn yield_now() {
-    wasip3::wit_bindgen::yield_async().await
+    wasip3::wit_bindgen::yield_async().await;
 }
 
 // Modern WASI P3 TcpSocket using wit_stream for async I/O
 pub struct TcpSocket {
     wasi_socket: WasiTcpSocket,
     read_buffer: BytesMut,
+    // Active read operation using WASI's waitable system
+    read_operation: Option<Pin<Box<dyn std::future::Future<Output = std::io::Result<Vec<u8>>> + Send + Sync>>>,
+    // Active write operation using WASI's waitable system
+    // write_operation: Option<Pin<Box<dyn std::future::Future<Output = std::io::Result<usize>> + Send + Sync>>>,
+    // // Write buffer for pending data
+    // write_buffer: BytesMut,
+    // Write readiness state
+    write_ready: bool,
 }
 
 impl TcpSocket {
     fn new(wasi_socket: WasiTcpSocket) -> Self {
         Self {
             wasi_socket,
             read_buffer: BytesMut::new(),
+            read_operation: None,
+            // write_operation: None,
+            // write_buffer: BytesMut::new(),
+            write_ready: true,
         }
     }
 
@@ -82,36 +91,40 @@ impl TcpSocket {
             (StreamResult::Complete(_), data) => {
                 let to_copy = std::cmp::min(buf.len(), data.len());
                 buf[..to_copy].copy_from_slice(&data[..to_copy]);
-                
+
                 // Buffer remaining data
                 if data.len() > to_copy {
                     self.read_buffer.extend_from_slice(&data[to_copy..]);
                 }
-                
+
                 Ok(to_copy)
             }
             (StreamResult::Dropped, _) => Ok(0),
-            (StreamResult::Cancelled, _) => {
-                Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "Read cancelled"))
-            }
+            (StreamResult::Cancelled, _) => Err(std::io::Error::new(
+                std::io::ErrorKind::Interrupted,
+                "Read cancelled",
+            )),
         }
     }
 
     pub async fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         let (mut tx, rx) = wit_stream::new();
-        
+
         // Start the send operation asynchronously
         let send_fut = self.wasi_socket.send(rx);
-        
+
         // Write the data
         let remaining = tx.write_all(buf.to_vec()).await;
         drop(tx);
-        
+
         // Wait for send to complete
         send_fut.await.map_err(|e| {
-            std::io::Error::new(std::io::ErrorKind::BrokenPipe, format!("Send failed: {:?}", e))
+            std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                format!("Send failed: {:?}", e),
+            )
         })?;
-        
+
         if remaining.is_empty() {
             Ok(buf.len())
         } else {
@@ -125,12 +138,22 @@ pub async fn connect_tcp<Ws: WithSocket>(
     port: u16,
     with_socket: Ws,
 ) -> crate::Result<Ws::Output> {
-    let addresses = wasip3::sockets::ip_name_lookup::resolve_addresses(host.to_string()).await
-        .map_err(|e| crate::Error::Io(std::io::Error::new(std::io::ErrorKind::Other, format!("DNS failed: {:?}", e))))?;
-    
-    let ip = addresses.into_iter().next()
-        .ok_or_else(|| crate::Error::Io(std::io::Error::new(std::io::ErrorKind::Other, "No addresses found")))?;
-    
+    let addresses = wasip3::sockets::ip_name_lookup::resolve_addresses(host.to_string())
+        .await
+        .map_err(|e| {
+            crate::Error::Io(std::io::Error::new(
+                std::io::ErrorKind::Other,
+                format!("DNS failed: {:?}", e),
+            ))
+        })?;
+
+    let ip = addresses.into_iter().next().ok_or_else(|| {
+        crate::Error::Io(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "No addresses found",
+        ))
+    })?;
+
     let addr = match ip {
         wasip3::sockets::types::IpAddress::Ipv4(ipv4) => {
             IpSocketAddress::Ipv4(wasip3::sockets::types::Ipv4SocketAddress {
@@ -154,7 +177,7 @@ pub async fn connect_tcp<Ws: WithSocket>(
             format!("failed to create socket: {:?}", e),
         ))
     })?;
-    
+
     wasi_socket.connect(addr).await.map_err(|e| {
         crate::Error::Io(std::io::Error::new(
             std::io::ErrorKind::ConnectionRefused,
@@ -165,4 +188,4 @@ pub async fn connect_tcp<Ws: WithSocket>(
     let tcp_socket = TcpSocket::new(wasi_socket);
 
     Ok(with_socket.with_socket(tcp_socket).await)
-}
+}