fix(wasm): avoid std::time::Instant::now() and expand testing

Cargo.toml

@@ -44,6 +44,7 @@ thiserror = "2.0.2"
 tracing = { version = "0.1" }
 url = "2.2"
 walkdir = { version = "2", optional = true }
+web-time = { version = "1.1.0" }
 
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
@@ -61,13 +62,19 @@ serde = { version = "1.0", default-features = false, features = ["derive"], opti
 serde_json = { version = "1.0", default-features = false, features = ["std"], optional = true }
 serde_urlencoded = { version = "0.7", optional = true }
 tokio = { version = "1.29.0", features = ["sync", "macros", "rt", "time", "io-util"] }
+n0-future = "0.3.2"
 
 [target.'cfg(target_family="unix")'.dev-dependencies]
 nix = { version = "0.31.1", features = ["fs"] }
 
 [target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies]
-web-time = { version = "1.1.0" }
+
 wasm-bindgen-futures = "0.4.18"
+send_wrapper = { version = "0.6", features = ["futures"] }
+gloo-timers = { version = "0.3", features = ["futures"] }
+chrono = { version = "0.4.34", default-features = false, features = ["clock", "wasmbind"] }
+
+
 
 [features]
 default = ["fs"]
@@ -89,7 +96,7 @@ regex = "1.11.1"
 # The "gzip" feature for reqwest is enabled for an integration test.
 reqwest = { version = "0.12", default-features = false, features = ["gzip"] }
 
-[target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dev-dependencies]
+[target.'cfg(target_arch = "wasm32")'.dev-dependencies]
 wasm-bindgen-test = "0.3.50"
 
 [dev-dependencies.getrandom_v03]

src/attributes.rs

@@ -197,6 +197,7 @@ impl<'a> Iterator for AttributesIter<'a> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::test;
 
     #[test]
     fn test_attributes_basic() {

src/buffered.rs

@@ -494,7 +494,9 @@ mod tests {
     use itertools::Itertools;
     use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 
-    #[tokio::test]
+    use crate::test_macros::*;
+
+    #[async_test]
     async fn test_buf_reader() {
         let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
 
@@ -574,7 +576,7 @@ mod tests {
     }
 
     // Note: `BufWriter::with_tags` functionality is tested in `crate::tests::tagging`
-    #[tokio::test]
+    #[async_test]
     async fn test_buf_writer() {
         let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
         let path = Path::from("file.txt");
@@ -612,7 +614,7 @@ mod tests {
         assert_eq!(response.attributes, attributes);
     }
 
-    #[tokio::test]
+    #[async_test]
     async fn test_buf_writer_with_put() {
         let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
         let path = Path::from("file.txt");

src/chunked.rs

@@ -182,10 +182,11 @@ mod tests {
     use crate::local::LocalFileSystem;
     use crate::memory::InMemory;
     use crate::path::Path;
+    use crate::test_macros::*;
 
     use super::*;
 
-    #[tokio::test]
+    #[async_test]
     async fn test_chunked_basic() {
         let location = Path::parse("test").unwrap();
         let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());

src/client/backoff.rs

@@ -110,6 +110,7 @@ impl Backoff {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::test;
     use rand::rand_core::impls::fill_bytes_via_next;
 
     struct FixedRng(u64);

src/client/builder.rs

@@ -272,6 +272,7 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::test;
 
     #[test]
     fn test_add_query_pairs() {

src/client/get.rs

@@ -222,7 +222,7 @@ impl<T: GetClient> GetContext<T> {
                                 sleep.as_secs_f32()
                             );
 
-                            tokio::time::sleep(sleep).await;
+                            crate::util::sleep(sleep).await;
 
                             let options = GetOptions {
                                 range: Some(GetRange::Bounded(range.clone())),
@@ -372,6 +372,7 @@ fn get_attributes(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::{async_test, test};
     use http::header::*;
 
     fn make_response(
@@ -407,7 +408,7 @@ mod tests {
         user_defined_metadata_prefix: Some("x-test-meta-"),
     };
 
-    #[tokio::test]
+    #[async_test]
     async fn test_get_range_meta() {
         let path = Path::from("test");
 

src/client/mod.rs

@@ -989,6 +989,7 @@ pub(crate) use cloud::*;
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::test;
     use std::collections::HashMap;
 
     #[test]

src/client/retry.rs

@@ -386,7 +386,7 @@ impl RetryableRequest {
                                 ctx.retries,
                                 ctx.max_retries,
                             );
-                            tokio::time::sleep(sleep).await;
+                            crate::util::sleep(sleep).await;
                         }
                     } else if status == StatusCode::NOT_MODIFIED {
                         return Err(self.err(RequestError::Status { status, body: None }, ctx));
@@ -431,7 +431,7 @@ impl RetryableRequest {
                             ctx.retries,
                             ctx.max_retries,
                         );
-                        tokio::time::sleep(sleep).await;
+                        crate::util::sleep(sleep).await;
                     }
                 }
                 Err(e) => {
@@ -457,7 +457,7 @@ impl RetryableRequest {
                         ctx.max_retries,
                         e,
                     );
-                    tokio::time::sleep(sleep).await;
+                    crate::util::sleep(sleep).await;
                 }
             }
         }
@@ -749,14 +749,14 @@ mod tests {
 
         // Retries on client timeout
         mock.push_async_fn(|_| async move {
-            tokio::time::sleep(Duration::from_secs(10)).await;
+            crate::util::sleep(Duration::from_secs(10)).await;
             panic!()
         });
         do_request().await.unwrap();
 
         // Does not retry PUT request
         mock.push_async_fn(|_| async move {
-            tokio::time::sleep(Duration::from_secs(10)).await;
+            crate::util::sleep(Duration::from_secs(10)).await;
             panic!()
         });
         let res = client.request(Method::PUT, mock.url()).send_retry(&retry);

src/client/token.rs

@@ -16,8 +16,8 @@
 // under the License.
 
 use std::future::Future;
-use std::time::{Duration, Instant};
 use tokio::sync::Mutex;
+use web_time::{Duration, Instant};
 
 /// A temporary authentication token with an associated expiry
 #[derive(Debug, Clone)]
@@ -91,6 +91,7 @@ impl<T: Clone + Send> TokenCache<T> {
 #[cfg(test)]
 mod test {
     use crate::client::token::{TemporaryToken, TokenCache};
+    use crate::test_macros::async_test;
     use std::sync::atomic::{AtomicU32, Ordering};
     use std::time::{Duration, Instant};
 
@@ -102,7 +103,7 @@ mod test {
         }
     }
 
-    #[tokio::test]
+    #[async_test]
     async fn test_expired_token_is_refreshed() {
         let cache = TokenCache::default();
         static COUNTER: AtomicU32 = AtomicU32::new(0);
@@ -116,14 +117,14 @@ mod test {
         let _ = cache.get_or_insert_with(get_token).await.unwrap();
         assert_eq!(COUNTER.load(Ordering::SeqCst), 1);
 
-        tokio::time::sleep(Duration::from_millis(2)).await;
+        crate::util::sleep(Duration::from_millis(2)).await;
 
         // Token is expired, so should fetch again
         let _ = cache.get_or_insert_with(get_token).await.unwrap();
         assert_eq!(COUNTER.load(Ordering::SeqCst), 2);
     }
 
-    #[tokio::test]
+    #[async_test]
     async fn test_min_ttl_causes_refresh() {
         let cache = TokenCache {
             cache: Default::default(),
@@ -146,7 +147,7 @@ mod test {
         let _ = cache.get_or_insert_with(get_token).await.unwrap();
         assert_eq!(COUNTER.load(Ordering::SeqCst), 1);
 
-        tokio::time::sleep(Duration::from_millis(2)).await;
+        crate::util::sleep(Duration::from_millis(2)).await;
 
         // Should fetch, since we've passed fetch_backoff
         let _ = cache.get_or_insert_with(get_token).await.unwrap();

src/config.rs

@@ -131,6 +131,7 @@ pub(crate) fn fmt_duration(duration: &ConfigValue<Duration>) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::test;
     use std::time::Duration;
 
     #[test]

src/delimited.rs

@@ -184,6 +184,7 @@ where
 
 #[cfg(test)]
 mod tests {
+    use crate::test_macros::{async_test, test};
     use futures::stream::{BoxStream, TryStreamExt};
 
     use super::*;
@@ -228,7 +229,7 @@ mod tests {
         assert!(delimiter.next().is_none());
     }
 
-    #[tokio::test]
+    #[async_test]
     async fn test_delimiter_stream() {
         let input = vec!["hello\nworld\nbin", "go\ncup", "cakes"];
         let input_stream = futures::stream::iter(input.into_iter().map(|s| Ok(Bytes::from(s))));
@@ -244,7 +245,7 @@ mod tests {
             ]
         )
     }
-    #[tokio::test]
+    #[async_test]
     async fn test_delimiter_unfold_stream() {
         let input_stream: BoxStream<'static, Result<Bytes>> = futures::stream::unfold(
             VecDeque::from(["hello\nworld\nbin", "go\ncup", "cakes"]),

src/lib.rs

@@ -2036,6 +2036,7 @@ pub enum Error {
         source: path::Error,
     },
 
+    #[cfg(not(target_arch = "wasm32"))]
     /// Error when `tokio::spawn` failed
     #[error("Error joining spawned task: {}", source)]
     JoinError {
@@ -2044,6 +2045,15 @@ pub enum Error {
         source: tokio::task::JoinError,
     },
 
+    #[cfg(target_arch = "wasm32")]
+    /// Error when `tokio::spawn` failed
+    #[error("Error joining spawned task: {}", source)]
+    JoinError {
+        /// The wrapped error
+        #[from]
+        source: n0_future::task::JoinError,
+    },
+
     /// Error when the attempted operation is not supported
     #[error("Operation not supported: {}", source)]
     NotSupported {
@@ -2138,9 +2148,26 @@ impl From<Error> for std::io::Error {
     }
 }
 
+/// Configure the test macro to use
+#[cfg(test)]
+pub mod test_macros {
+    #[cfg(not(target_arch = "wasm32"))]
+    pub use test;
+
+    #[cfg(target_arch = "wasm32")]
+    pub use wasm_bindgen_test::wasm_bindgen_test as test;
+
+    #[cfg(not(target_arch = "wasm32"))]
+    pub use tokio::test as async_test;
+
+    #[cfg(target_arch = "wasm32")]
+    pub use wasm_bindgen_test::wasm_bindgen_test as async_test;
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_macros::{async_test, test};
 
     use chrono::TimeZone;
 
@@ -2267,7 +2294,7 @@ mod tests {
         }
     }
 
-    #[tokio::test]
+    #[async_test]
     async fn test_list_lifetimes() {
         let store = memory::InMemory::new();
         let mut stream = list_store(&store, "path");

src/limit.rs

@@ -247,10 +247,10 @@ mod tests {
     use crate::memory::InMemory;
     use futures::stream::StreamExt;
     use std::pin::Pin;
-    use std::time::Duration;
-    use tokio::time::timeout;
 
-    #[tokio::test]
+    use crate::test_macros::*;
+
+    #[async_test]
     async fn limit_test() {
         let max_requests = 10;
         let memory = InMemory::new();
@@ -270,11 +270,14 @@ mod tests {
             streams.push(stream);
         }
 
-        let t = Duration::from_millis(20);
-
         // Expect to not be able to make another request
-        let fut = integration.list(None).collect::<Vec<_>>();
-        assert!(timeout(t, fut).await.is_err());
+        #[cfg(not(target_arch = "wasm32"))]
+        {
+            // tokio timeout will panic on WASM
+            let t = std::time::Duration::from_millis(20);
+            let fut = integration.list(None).collect::<Vec<_>>();
+            assert!(tokio::time::timeout(t, fut).await.is_err());
+        }
 
         // Drop one of the streams
         streams.pop();