refactor: build on wasm, do not use send_blocking

Author: Frando

Cargo.lock

@@ -37,7 +37,7 @@ iroh-blobs = { git = "https://github.com/n0-computer/iroh-blobs", branch = "Fran
 iroh-gossip = { version = "0.94", features = ["net"], default-features = false }
 iroh-metrics = { version = "0.36", default-features = false }
 irpc = { git = "https://github.com/n0-computer/irpc", branch = "main", default-features = false }
-n0-future = "0.3"
+n0-future = { version = "0.3", features = ["serde"] }
 num_enum = "0.7"
 postcard = { version = "1", default-features = false, features = [
     "alloc",
@@ -98,3 +98,6 @@ unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)"] }
 
 [build-dependencies]
 cfg_aliases = "0.2.1"
+
+[patch.crates-io]
+n0-future = { path = "../n0-future" }

Cargo.toml

@@ -4,7 +4,6 @@ use std::{
     collections::{hash_map, HashMap},
     num::NonZeroU64,
     sync::Arc,
-    thread::JoinHandle,
     time::Duration,
 };
 
@@ -18,6 +17,9 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::oneshot;
 use tracing::{debug, error, error_span, trace, warn};
 
+#[cfg(wasm_browser)]
+use tracing::Instrument;
+
 use crate::{
     api::{
         protocol::{AuthorListResponse, ListResponse},
@@ -228,6 +230,7 @@ struct OpenReplica {
 pub struct SyncHandle {
     tx: async_channel::Sender<Action>,
     #[cfg(wasm_browser)]
+    #[allow(unused)]
     join_handle: Arc<Option<n0_future::task::JoinHandle<()>>>,
     #[cfg(not(wasm_browser))]
     join_handle: Arc<Option<std::thread::JoinHandle<()>>>,
@@ -275,21 +278,22 @@ impl SyncHandle {
             metrics: metrics.clone(),
         };
 
+        let span = error_span!("sync", %me);
         #[cfg(wasm_browser)]
-        let join_handle = n0_future::task::spawn(actor.run_async());
+        let join_handle = n0_future::task::spawn(actor.run_async().instrument(span));
 
         #[cfg(not(wasm_browser))]
         let join_handle = std::thread::Builder::new()
             .name("sync-actor".to_string())
             .spawn(move || {
-                let span = error_span!("sync", %me);
                 let _enter = span.enter();
 
                 if let Err(err) = actor.run_in_thread() {
                     error!("Sync actor closed with error: {err:?}");
                 }
             })
             .expect("failed to spawn thread");
+
         let join_handle = Arc::new(Some(join_handle));
         SyncHandle {
             tx: action_tx,
@@ -602,15 +606,22 @@ impl SyncHandle {
 
 impl Drop for SyncHandle {
     fn drop(&mut self) {
+        #[cfg(wasm_browser)]
+        {
+            let tx = self.tx.clone();
+            n0_future::task::spawn(async move {
+                tx.send(Action::Shutdown { reply: None }).await.ok();
+            });
+        }
         // this means we're dropping the last reference
+        #[cfg(not(wasm_browser))]
         if let Some(handle) = Arc::get_mut(&mut self.join_handle) {
             // this call is the reason tx can not be a tokio mpsc channel.
             // we have no control about where drop is called, yet tokio send_blocking panics
             // when called from inside a tokio runtime.
             self.tx.send_blocking(Action::Shutdown { reply: None }).ok();
             let handle = handle.take().expect("this can only run once");
 
-            #[cfg(not(wasm_browser))]
             if let Err(err) = handle.join() {
                 warn!(?err, "Failed to join sync actor");
             }
@@ -628,10 +639,7 @@ struct Actor {
 }
 
 impl Actor {
-    async fn run_in_task(self) {
-        self.run_async().await
-    }
-
+    #[cfg(not(wasm_browser))]
     fn run_in_thread(self) -> Result<()> {
         let rt = tokio::runtime::Builder::new_current_thread()
             .enable_time()
@@ -779,37 +787,46 @@ impl Actor {
                 hash,
                 len,
                 reply,
-            } => send_reply_with(reply, self, move |this| {
-                let author = get_author(&mut this.store, &author)?;
-                let mut replica = this.states.replica(namespace, &mut this.store)?;
-                replica.insert(&key, &author, hash, len)?;
-                this.metrics.new_entries_local.inc();
-                this.metrics.new_entries_local_size.inc_by(len);
-                Ok(())
-            }),
+            } => {
+                send_reply_with_async(reply, self, async move |this| {
+                    let author = get_author(&mut this.store, &author)?;
+                    let mut replica = this.states.replica(namespace, &mut this.store)?;
+                    replica.insert(&key, &author, hash, len).await?;
+                    this.metrics.new_entries_local.inc();
+                    this.metrics.new_entries_local_size.inc_by(len);
+                    Ok(())
+                })
+                .await
+            }
             ReplicaAction::DeletePrefix { author, key, reply } => {
-                send_reply_with(reply, self, |this| {
+                send_reply_with_async(reply, self, async |this| {
                     let author = get_author(&mut this.store, &author)?;
                     let mut replica = this.states.replica(namespace, &mut this.store)?;
-                    let res = replica.delete_prefix(&key, &author)?;
+                    let res = replica.delete_prefix(&key, &author).await?;
                     Ok(res)
                 })
+                .await
             }
             ReplicaAction::InsertRemote {
                 entry,
                 from,
                 content_status,
                 reply,
-            } => send_reply_with(reply, self, move |this| {
-                let mut replica = this
-                    .states
-                    .replica_if_syncing(&namespace, &mut this.store)?;
-                let len = entry.content_len();
-                replica.insert_remote_entry(entry, from, content_status)?;
-                this.metrics.new_entries_remote.inc();
-                this.metrics.new_entries_remote_size.inc_by(len);
-                Ok(())
-            }),
+            } => {
+                send_reply_with_async(reply, self, async move |this| {
+                    let mut replica = this
+                        .states
+                        .replica_if_syncing(&namespace, &mut this.store)?;
+                    let len = entry.content_len();
+                    replica
+                        .insert_remote_entry(entry, from, content_status)
+                        .await?;
+                    this.metrics.new_entries_remote.inc();
+                    this.metrics.new_entries_remote_size.inc_by(len);
+                    Ok(())
+                })
+                .await
+            }
 
             ReplicaAction::SyncInitialMessage { reply } => {
                 send_reply_with(reply, self, move |this| {
@@ -1064,6 +1081,14 @@ fn send_reply_with<T>(
     sender.send(f(this)).map_err(send_reply_error)
 }
 
+async fn send_reply_with_async<T>(
+    sender: oneshot::Sender<Result<T>>,
+    this: &mut Actor,
+    f: impl AsyncFnOnce(&mut Actor) -> Result<T>,
+) -> Result<(), SendReplyError> {
+    sender.send(f(this).await).map_err(send_reply_error)
+}
+
 fn send_reply_error<T>(_err: T) -> SendReplyError {
     SendReplyError
 }

src/actor.rs

@@ -429,7 +429,7 @@ impl RpcActor {
                 return;
             }
         };
-        tokio::task::spawn(async move {
+        n0_future::task::spawn(async move {
             loop {
                 tokio::select! {
                     msg = stream.next() => {

src/api/actor.rs

@@ -2,13 +2,9 @@
 //!
 //! [`crate::Replica`] is also called documents here.
 
-use std::{
-    path::PathBuf,
-    str::FromStr,
-    sync::{Arc, RwLock},
-};
+use std::sync::{Arc, RwLock};
 
-use anyhow::{bail, Context, Result};
+use anyhow::{bail, Result};
 use futures_lite::{Stream, StreamExt};
 use iroh::{Endpoint, EndpointAddr, PublicKey};
 use iroh_blobs::{
@@ -17,9 +13,9 @@ use iroh_blobs::{
     Hash,
 };
 use iroh_gossip::net::Gossip;
+use n0_future::task::AbortOnDropHandle;
 use serde::{Deserialize, Serialize};
 use tokio::sync::{mpsc, oneshot};
-use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, error_span, Instrument};
 
 use self::live::{LiveActor, ToLiveActor};
@@ -128,7 +124,7 @@ impl Engine {
             live_actor_tx.clone(),
             sync.metrics().clone(),
         );
-        let actor_handle = tokio::task::spawn(
+        let actor_handle = n0_future::task::spawn(
             async move {
                 if let Err(err) = actor.run().await {
                     error!("sync actor failed: {err:?}");
@@ -355,7 +351,8 @@ pub enum DefaultAuthorStorage {
     /// Memory storage.
     Mem,
     /// File based persistent storage.
-    Persistent(PathBuf),
+    #[cfg(feature = "fs-store")]
+    Persistent(std::path::PathBuf),
 }
 
 impl DefaultAuthorStorage {
@@ -373,7 +370,10 @@ impl DefaultAuthorStorage {
                 docs_store.import_author(author).await?;
                 Ok(author_id)
             }
+            #[cfg(feature = "fs-store")]
             Self::Persistent(ref path) => {
+                use anyhow::Context;
+                use std::str::FromStr;
                 if path.exists() {
                     let data = tokio::fs::read_to_string(path).await.with_context(|| {
                         format!(
@@ -407,12 +407,14 @@ impl DefaultAuthorStorage {
     }
 
     /// Save a new default author.
-    pub async fn persist(&self, author_id: AuthorId) -> anyhow::Result<()> {
+    pub async fn persist(&self, #[allow(unused)] author_id: AuthorId) -> anyhow::Result<()> {
         match self {
             Self::Mem => {
                 // persistence is not possible for the mem storage so this is a noop.
             }
+            #[cfg(feature = "fs-store")]
             Self::Persistent(ref path) => {
+                use anyhow::Context;
                 tokio::fs::write(path, author_id.to_string())
                     .await
                     .with_context(|| {

src/engine.rs

@@ -3,8 +3,8 @@
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
-    time::SystemTime,
 };
+use n0_future::time::SystemTime;
 
 use anyhow::{Context, Result};
 use futures_lite::FutureExt;

src/engine/live.rs

@@ -1,7 +1,8 @@
 use std::{
     collections::BTreeMap,
-    time::{Instant, SystemTime},
+    time::Instant,
 };
+use n0_future::time::SystemTime;
 
 use anyhow::Result;
 use iroh::EndpointId;

src/engine/state.rs

@@ -322,13 +322,15 @@ mod tests {
         let alice_replica_id = alice_replica.id();
         alice_replica
             .hash_and_insert("hello bob", &author, "from alice")
+            .await
             .unwrap();
 
         let mut bob_store = store::Store::memory();
         let mut bob_replica = bob_store.new_replica(namespace.clone()).unwrap();
         let bob_replica_id = bob_replica.id();
         bob_replica
             .hash_and_insert("hello alice", &author, "from bob")
+            .await
             .unwrap();
 
         assert_eq!(
@@ -438,9 +440,9 @@ mod tests {
 
     type Message = (AuthorId, Vec<u8>, Hash);
 
-    fn insert_messages(
+    async fn insert_messages(
         mut rng: impl CryptoRng,
-        replica: &mut crate::sync::Replica,
+        replica: &mut crate::sync::Replica<'_>,
         num_authors: usize,
         msgs_per_author: usize,
         key_value_fn: impl Fn(&AuthorId, usize) -> (String, String),
@@ -453,7 +455,10 @@ mod tests {
         for i in 0..msgs_per_author {
             for author in authors.iter() {
                 let (key, value) = key_value_fn(&author.id(), i);
-                let hash = replica.hash_and_insert(key.clone(), author, value).unwrap();
+                let hash = replica
+                    .hash_and_insert(key.clone(), author, value)
+                    .await
+                    .unwrap();
                 res.push((author.id(), key.as_bytes().to_vec(), hash));
             }
         }
@@ -509,7 +514,8 @@ mod tests {
                             format!("from alice by {author}: {i}"),
                         )
                     },
-                );
+                )
+                .await;
                 all_messages.extend_from_slice(&alice_messages);
 
                 let mut bob_replica = bob_store.new_replica(namespace.clone()).unwrap();
@@ -524,7 +530,8 @@ mod tests {
                             format!("from bob by {author}: {i}"),
                         )
                     },
-                );
+                )
+                .await;
                 all_messages.extend_from_slice(&bob_messages);
 
                 all_messages.sort();
@@ -646,10 +653,12 @@ mod tests {
         // Insert into alice
         let hash_alice = alice_replica
             .hash_and_insert(&key, &author, &value_alice)
+            .await
             .unwrap();
         // Insert into bob
         let hash_bob = bob_replica
             .hash_and_insert(&key, &author, &value_bob)
+            .await
             .unwrap();
 
         assert_eq!(