macltok

public

Star0

Fork 0

Watch0

6 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `1fe480c16a3c`

tarball

atagen <boss@atagen.co>2026-09-27 01:42 UTC

10 files changed+194-99

engine: make the executor portable

Spawn, sleep and timeout go through matrix_sdk's executor, which is tokio natively and spawn_local with gloo timers on wasm, so the engine no longer assumes tokio's runtime. TaskSet stands in for JoinSet and aborts what's left on drop like JoinSet did, or a dead room actor would leave its redactions running and their callers would never hear Interrupted. Tasks catch their own panics so each comes back with its context, since joining through the sdk executor gives no task id to look it up by, and upload forwarders are pruned by their closed ready channel since the wasm JoinHandle can't tell a finished task from a live one.

Modified file: Cargo.toml+1−1

@@ -21,7 +21,7 @@ serde = { version = "1.0.219", features = ["derive", "rc"], optional = true }

2121 thiserror = "2.0.16"

2222 mime = "0.3.17"

2323 sha2 = "0.10.9"

24-tokio = { version = "1.47.1", features = ["rt", "sync", "time"] }

24+tokio = { version = "1.47.1", features = ["macros", "sync"] }

2525 tracing = "0.1.41"

2626 url = { version = "2.5.7", optional = true }

2727

Modified file: src/encryption.rs+4 additions−3 deletions@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use futures_util::StreamExt;
+use matrix_sdk::executor::{JoinHandle, spawn};
 use matrix_sdk::{
     Client,
     encryption::{
@@ -14,7 +15,7 @@ use matrix_sdk::{
     ruma::events::key::verification::VerificationMethod,
 };
 use ruma::{OwnedDeviceId, UserId};
-use tokio::{sync::Mutex, task::JoinHandle};
+use tokio::sync::Mutex;
 
 use crate::{
     EncryptionState, Error, ErrorKind, Event, KeyBackupState, RecoveryState, VerificationAction,
@@ -266,7 +267,7 @@ impl VerificationController {
                 Error::with_source(ErrorKind::Interrupted, "event receiver has closed", source)
             })?;
         let active = self.active.clone();
-        self.task = Some(tokio::spawn(async move {
+        self.task = Some(spawn(async move {
             follow_request(request, changes, active, sender).await;
         }));
         Ok(())
@@ -311,7 +312,7 @@ impl VerificationController {
             }
         }
         let active = self.active.clone();
-        self.task = Some(tokio::spawn(async move {
+        self.task = Some(spawn(async move {
             follow_request(request, changes, active, sender).await;
         }));
     }

Added file: src/engine/executor.rs+82 additions@@ -0,0 +1,82 @@
+use std::future::Future;
+
+use futures_util::{StreamExt, stream::FuturesUnordered};
+use matrix_sdk::SendOutsideWasm;
+pub(super) use matrix_sdk::executor::{AbortHandle, JoinError, JoinHandle, spawn};
+
+#[cfg(not(target_family = "wasm"))]
+pub(super) type EngineBoxFuture<'a, T> = futures_util::future::BoxFuture<'a, T>;
+#[cfg(target_family = "wasm")]
+pub(super) type EngineBoxFuture<'a, T> = futures_util::future::LocalBoxFuture<'a, T>;
+
+#[cfg(not(target_family = "wasm"))]
+pub(super) fn is_panic(error: &JoinError) -> bool {
+    error.is_panic()
+}
+
+/// the sdk's wasm join error has no payload, though a panic there aborts the instance anyway
+#[cfg(target_family = "wasm")]
+pub(super) fn is_panic(error: &JoinError) -> bool {
+    !error.is_cancelled()
+}
+
+#[cfg(not(target_family = "wasm"))]
+pub(super) fn resume_unwind(error: JoinError) -> ! {
+    std::panic::resume_unwind(error.into_panic())
+}
+
+#[cfg(target_family = "wasm")]
+pub(super) fn resume_unwind(error: JoinError) -> ! {
+    panic!("{error}")
+}
+
+/// a `JoinSet` that runs on wasm too, and like it aborts whatever's left when dropped
+pub(super) struct TaskSet<T: 'static> {
+    handles: FuturesUnordered<JoinHandle<T>>,
+}
+
+impl<T: 'static> TaskSet<T> {
+    pub(super) fn new() -> Self {
+        Self {
+            handles: FuturesUnordered::new(),
+        }
+    }
+
+    pub(super) fn spawn(
+        &mut self,
+        task: impl Future<Output = T> + SendOutsideWasm + 'static,
+    ) -> AbortHandle
+    where
+        T: SendOutsideWasm,
+    {
+        let handle = spawn(task);
+        let abort = handle.abort_handle();
+        self.handles.push(handle);
+        abort
+    }
+
+    pub(super) fn is_empty(&self) -> bool {
+        self.handles.is_empty()
+    }
+
+    pub(super) async fn join_next(&mut self) -> Option<Result<T, JoinError>> {
+        self.handles.next().await
+    }
+
+    fn abort_all(&self) {
+        for handle in &self.handles {
+            handle.abort();
+        }
+    }
+
+    pub(super) async fn shutdown(&mut self) {
+        self.abort_all();
+        while self.handles.next().await.is_some() {}
+    }
+}
+
+impl<T: 'static> Drop for TaskSet<T> {
+    fn drop(&mut self) {
+        self.abort_all();
+    }
+}

Modified file: src/engine/media.rs+15 additions−10 deletions@@ -9,10 +9,10 @@ use std::{
 
 use futures_util::{
     FutureExt, StreamExt,
-    future::{BoxFuture, Shared, WeakShared},
+    future::{Shared, WeakShared},
 };
 use matrix_sdk::{
-    Client, Room,
+    Client, Room, SendOutsideWasm, SyncOutsideWasm,
     media::{MediaFormat, MediaRequestParameters, MediaThumbnailSettings},
     room::reply::{EnforceThread, Reply as SdkReply},
     ruma::{
@@ -31,6 +31,7 @@ use matrix_sdk_crypto::{AttachmentDecryptor, MediaEncryptionInfo};
 use ruma::MxcUri;
 use tokio::sync::Mutex;
 
+use super::executor::EngineBoxFuture;
 use crate::{
     AttachmentUpload, Error, ErrorKind, Media, MediaId, MediaKind, MediaSize, TimelineId,
     normalise::{MediaRegistration, MediaScope},
@@ -42,7 +43,7 @@ const DOWNLOAD_IDLE: Duration = Duration::from_secs(60);
 /// the sdk client gives up on any request after thirty seconds, far too soon for a video
 const DOWNLOAD_TOTAL: Duration = Duration::from_secs(60 * 60);
 
-type FlightFuture = BoxFuture<'static, Result<Media, Error>>;
+type FlightFuture = EngineBoxFuture<'static, Result<Media, Error>>;
 type Flight = Shared<FlightFuture>;
 
 /// owns everything needed to turn a media id or an mxc uri into bytes, so the
@@ -188,7 +189,7 @@ impl MediaService {
             match flights.get(&key).and_then(WeakShared::upgrade) {
                 Some(flight) => flight,
                 None => {
-                    let flight: Flight = AssertUnwindSafe(fetch(
+                    let prepared = AssertUnwindSafe(fetch(
                         self.client.clone(),
                         request,
                         mimetype,
@@ -199,9 +200,9 @@ impl MediaService {
                         result.unwrap_or_else(|_| {
                             Err(Error::new(ErrorKind::Other, "media fetch panicked"))
                         })
-                    })
-                    .boxed()
-                    .shared();
+                    });
+                    let boxed: FlightFuture = Box::pin(prepared);
+                    let flight: Flight = boxed.shared();
                     if let Some(weak) = flight.downgrade() {
                         flights.insert(key, weak);
                     }
@@ -335,7 +336,9 @@ async fn cached_media(
     client: &Client,
     request: &MediaRequestParameters,
 ) -> Result<Option<Vec<u8>>, Error> {
-    fn cache_failed(source: impl std::error::Error + Send + Sync + 'static) -> Error {
+    fn cache_failed(
+        source: impl std::error::Error + SendOutsideWasm + SyncOutsideWasm + 'static,
+    ) -> Error {
         Error::with_source(ErrorKind::Store, "reading media cache failed", source)
     }
     client
@@ -356,7 +359,9 @@ fn maybe_decrypt(
     let MediaSource::Encrypted(file) = &request.source else {
         return Ok(bytes);
     };
-    fn decrypt_failed(source: impl std::error::Error + Send + Sync + 'static) -> Error {
+    fn decrypt_failed(
+        source: impl std::error::Error + SendOutsideWasm + SyncOutsideWasm + 'static,
+    ) -> Error {
         Error::with_source(
             ErrorKind::MediaUnavailable,
             "media decryption failed",
@@ -447,7 +452,7 @@ async fn download_bounded(
     let mut stream = response.bytes_stream();
     let mut bytes = Vec::new();
     loop {
-        let chunk = tokio::time::timeout(DOWNLOAD_IDLE, stream.next())
+        let chunk = matrix_sdk::timeout::timeout(stream.next(), DOWNLOAD_IDLE)
             .await
             .map_err(|_| Error::new(ErrorKind::Network, "media download stalled"))?;
         let Some(chunk) = chunk else { break };

Modified file: src/engine/mod.rs+51 additions−43 deletions@@ -1,4 +1,5 @@
 mod commands;
+mod executor;
 mod media;
 mod mutations;
 mod room;
@@ -8,23 +9,22 @@ mod watchers;
 use std::{
     collections::{HashMap, VecDeque},
     future::Future,
+    panic::AssertUnwindSafe,
     sync::Arc,
 };
 
-use futures_util::future::BoxFuture;
-use matrix_sdk::{Client, SessionChange};
+use futures_util::FutureExt;
+use matrix_sdk::{Client, SendOutsideWasm, SessionChange};
 use matrix_sdk_ui::sync_service::SyncService;
 use ruma::OwnedRoomId;
 use thiserror::Error as ThisError;
-use tokio::{
-    sync::{Notify, broadcast, mpsc, watch},
-    task::{AbortHandle, Id as TaskId, JoinError, JoinSet},
-};
+use tokio::sync::{Notify, broadcast, mpsc, watch};
 
 pub(crate) use media::MediaService;
 
 use self::{
     commands::Job,
+    executor::{EngineBoxFuture, TaskSet},
     mutations::{Mutation, MutationJob},
     room::{ActorNote, ActorRequest, RoomActor},
     timeline::{SessionContext, Supervisor},
@@ -143,36 +143,49 @@ impl TaskContext {
     }
 }
 
+/// tasks catch their own panics, since the sdk's wasm join error drops the payload
+fn panic_message(payload: Box<dyn std::any::Any + Send>) -> String {
+    payload
+        .downcast_ref::<&str>()
+        .map(ToString::to_string)
+        .or_else(|| payload.downcast_ref::<String>().cloned())
+        .unwrap_or_else(|| "task panicked with a non-string payload".to_owned())
+}
+
 struct TaskRegistry {
-    tasks: JoinSet<()>,
-    contexts: HashMap<TaskId, TaskContext>,
+    tasks: TaskSet<(TaskContext, Option<String>)>,
 }
 
 impl TaskRegistry {
     fn spawn(
         &mut self,
         context: TaskContext,
-        task: impl Future<Output = ()> + Send + 'static,
-    ) -> AbortHandle {
-        let handle = self.tasks.spawn(task);
-        self.contexts.insert(handle.id(), context);
-        handle
-    }
-
-    async fn join_next(&mut self) -> Option<(TaskContext, Result<(), JoinError>)> {
-        let result = self.tasks.join_next_with_id().await?;
-        let id = match &result {
-            Ok((id, ())) => *id,
-            Err(error) => error.id(),
-        };
-        let context = self.contexts.remove(&id)?;
-        Some((context, result.map(drop)))
+        task: impl Future<Output = ()> + SendOutsideWasm + 'static,
+    ) {
+        self.tasks.spawn(async move {
+            let panic = AssertUnwindSafe(task)
+                .catch_unwind()
+                .await
+                .err()
+                .map(panic_message);
+            (context, panic)
+        });
+    }
+
+    fn is_empty(&self) -> bool {
+        self.tasks.is_empty()
+    }
+
+    async fn join_next(&mut self) -> Option<(TaskContext, Option<String>)> {
+        loop {
+            if let Ok(done) = self.tasks.join_next().await? {
+                return Some(done);
+            }
+        }
     }
 
     async fn shutdown(&mut self) {
-        self.tasks.abort_all();
-        while self.tasks.join_next().await.is_some() {}
-        self.contexts.clear();
+        self.tasks.shutdown().await;
     }
 }
 
@@ -230,7 +243,7 @@ pub struct Engine {
     encryption: mpsc::Sender<Job<EncryptionCommand>>,
     queues: Option<Queues>,
     /// one progress forwarder per room that has uploaded, and whether it's subscribed yet
-    uploads: HashMap<OwnedRoomId, (AbortHandle, watch::Receiver<bool>)>,
+    uploads: HashMap<OwnedRoomId, watch::Receiver<bool>>,
     /// the client can outlive us, so the handler has to come off it when we stop
     verification_handler: Option<matrix_sdk::event_handler::EventHandlerHandle>,
     stop: Arc<Notify>,
@@ -283,8 +296,7 @@ impl Engine {
             config,
             media,
             tasks: TaskRegistry {
-                tasks: JoinSet::new(),
-                contexts: HashMap::new(),
+                tasks: TaskSet::new(),
             },
             rooms: HashMap::new(),
             next_generation: 0,
@@ -309,7 +321,7 @@ impl Engine {
 
     /// boxed because the engine's future is deep enough to blow the compiler's query depth
     /// in every crate that spawns it, which isn't something a client should have to fix
-    pub fn run(self) -> BoxFuture<'static, Result<(), EngineError>> {
+    pub fn run(self) -> EngineBoxFuture<'static, Result<(), EngineError>> {
         Box::pin(self.run_inner())
     }
 
@@ -346,13 +358,10 @@ impl Engine {
                     break;
                 }
                 () = self.stop.notified() => break,
-                Some((context, result)) = self.tasks.join_next(), if !self.tasks.tasks.is_empty() => {
-                    let Err(error) = result else { continue };
-                    if !error.is_panic() {
-                        continue;
-                    }
+                Some((context, panic)) = self.tasks.join_next(), if !self.tasks.is_empty() => {
+                    let Some(message) = panic else { continue };
                     if let TaskContext::Command(label) = context {
-                        tracing::error!(label, %error, "command task panicked");
+                        tracing::error!(label, panic = %message, "command task panicked");
                         continue;
                     }
                     failure = Some(EngineError::TaskPanicked { task: context.label() });
@@ -392,7 +401,7 @@ impl Engine {
         self.verification_handler = Some(self.verification_handler(incoming));
         let client = &self.client;
         let events = || self.port.event_sender();
-        let watchers: [(&'static str, BoxFuture<'static, ()>); 11] = [
+        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 11] = [
             (
                 "upload failure watcher",
                 Box::pin(watchers::upload_failures(
@@ -606,7 +615,7 @@ impl Engine {
         &mut self,
         label: &'static str,
         reply: Responder,
-        work: impl Future<Output = Result<(), Error>> + Send + 'static,
+        work: impl Future<Output = Result<(), Error>> + SendOutsideWasm + 'static,
     ) {
         self.tasks.spawn(TaskContext::Command(label), async move {
             reply.done(work.await);
@@ -630,18 +639,17 @@ impl Engine {
     }
 
     fn upload_forwarder(&mut self, room: &matrix_sdk::Room) -> watch::Receiver<bool> {
-        self.uploads
-            .retain(|_, (forwarder, _)| !forwarder.is_finished());
-        if let Some((_, ready)) = self.uploads.get(room.room_id()) {
+        self.uploads.retain(|_, ready| ready.has_changed().is_ok());
+        if let Some(ready) = self.uploads.get(room.room_id()) {
             return ready.clone();
         }
         let (subscribed, ready) = watch::channel(false);
-        let forwarder = self.tasks.spawn(
+        self.tasks.spawn(
             TaskContext::Command("upload watcher"),
             watchers::room_uploads(room.clone(), self.port.event_sender(), subscribed),
         );
         self.uploads
-            .insert(room.room_id().to_owned(), (forwarder, ready.clone()));
+            .insert(room.room_id().to_owned(), ready.clone());
         ready
     }
 

Modified file: src/engine/mutations.rs+4 additions−3 deletions@@ -19,9 +19,10 @@ use matrix_sdk::{
     },
 };
 use ruma::{OwnedRoomId, OwnedUserId};
-use tokio::{sync::mpsc, task::JoinSet};
+use tokio::sync::mpsc;
 
 use super::commands::{Job, guarded};
+use super::executor::TaskSet;
 use crate::{
     CommandOutput, CreateRoomOptions, Error, ErrorKind, PowerLevelChanges, RoomCommand, RoomPreset,
     RoomVisibility,
@@ -56,7 +57,7 @@ pub(super) async fn run_scheduler(
     let fresh_dms = FreshDms::default();
     let mut waiting = HashMap::<Key, VecDeque<MutationJob>>::new();
     let mut running = HashSet::<Key>::new();
-    let mut tasks = JoinSet::<Key>::new();
+    let mut tasks = TaskSet::<Key>::new();
     let mut queued = 0_usize;
     let mut independent = 0_u64;
     let mut receiving = true;
@@ -150,7 +151,7 @@ fn key(mutation: &Mutation, server: Option<&ruma::ServerName>) -> Option<Key> {
 type FreshDms = Arc<std::sync::Mutex<HashMap<OwnedUserId, OwnedRoomId>>>;
 
 fn spawn(
-    tasks: &mut JoinSet<Key>,
+    tasks: &mut TaskSet<Key>,
     client: &Client,
     fresh_dms: &FreshDms,
     key: Key,

Modified file: src/engine/room.rs+11 additions−14 deletions@@ -3,11 +3,9 @@ use std::collections::{HashMap, VecDeque};
 use matrix_sdk::{Room, RoomState};
 use matrix_sdk_ui::timeline::{RoomExt, Timeline, TimelineFocus, TimelineReadReceiptTracking};
 use ruma::{OwnedEventId, OwnedRoomId};
-use tokio::{
-    sync::mpsc,
-    task::{AbortHandle, JoinSet},
-};
+use tokio::sync::mpsc;
 
+use super::executor::{AbortHandle, JoinError, TaskSet, is_panic, resume_unwind};
 use super::timeline::{SessionContext, TaskGuard, TimelineSession};
 use crate::{
     Error, ErrorKind, Event, TimelineCommand, TimelineId, TypingState, transport::Responder,
@@ -100,7 +98,7 @@ impl RoomActor {
         let mut timelines = Timelines {
             room: room_timeline,
             threads: Threads::new(self.requests.max_capacity()),
-            work: JoinSet::new(),
+            work: TaskSet::new(),
         };
         let close = self.serve_open(&mut timelines).await;
         if close.is_none() {
@@ -150,7 +148,7 @@ impl RoomActor {
                 }
                 Some(label) = self.panics.recv() => panic!("{label} panicked in room {}", self.room_id),
                 Some(done) = timelines.work.join_next(), if !timelines.work.is_empty() => {
-                    if let Err(error) = done && error.is_panic() {
+                    if let Err(error) = done && is_panic(&error) {
                         tracing::warn!(room_id = %self.room_id, %error, "timeline operation panicked");
                     }
                 }
@@ -209,7 +207,7 @@ async fn watch_typing(
 struct Timelines {
     room: TimelineSession,
     threads: Threads,
-    work: JoinSet<()>,
+    work: TaskSet<()>,
 }
 
 impl Timelines {
@@ -262,7 +260,7 @@ impl Timelines {
 
     async fn thread_built(
         &mut self,
-        built: Result<Built, tokio::task::JoinError>,
+        built: Result<Built, JoinError>,
         room_id: &OwnedRoomId,
         context: &SessionContext,
     ) {
@@ -324,7 +322,7 @@ struct OpeningThread {
 struct Threads {
     open: HashMap<OwnedEventId, TimelineSession>,
     opening: HashMap<OwnedEventId, OpeningThread>,
-    builds: JoinSet<Built>,
+    builds: TaskSet<Built>,
     next_generation: u64,
     queued: usize,
     capacity: usize,
@@ -335,7 +333,7 @@ impl Threads {
         Self {
             open: HashMap::new(),
             opening: HashMap::new(),
-            builds: JoinSet::new(),
+            builds: TaskSet::new(),
             next_generation: 0,
             queued: 0,
             capacity,
@@ -403,7 +401,7 @@ impl Threads {
     /// `None` when the build belongs to a thread that was closed while it ran
     fn built(
         &mut self,
-        built: Result<Built, tokio::task::JoinError>,
+        built: Result<Built, JoinError>,
     ) -> Option<(
         OwnedEventId,
         Queued,
@@ -412,7 +410,7 @@ impl Threads {
         let (root, generation, timeline) = match built {
             Ok(built) => built,
             Err(error) if error.is_cancelled() => return None,
-            Err(error) => std::panic::resume_unwind(error.into_panic()),
+            Err(error) => resume_unwind(error),
         };
         if self
             .opening
@@ -443,8 +441,7 @@ impl Threads {
     }
 
     async fn close(&mut self) {
-        self.builds.abort_all();
-        while self.builds.join_next().await.is_some() {}
+        self.builds.shutdown().await;
         self.opening.clear();
         for (_, session) in self.open.drain() {
             session.close().await;

Modified file: src/engine/timeline.rs+10 additions−11 deletions@@ -1,6 +1,7 @@
-use std::{collections::HashSet, future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc};
+use std::{collections::HashSet, future::Future, panic::AssertUnwindSafe, sync::Arc};
 
 use futures_util::{FutureExt, StreamExt};
+use matrix_sdk::SendOutsideWasm;
 use matrix_sdk::{
     room::edit::EditedContent,
     room::reply::{EnforceThread, Reply as SdkReply},
@@ -16,11 +17,9 @@ use matrix_sdk::{
 };
 use matrix_sdk_ui::timeline::{Timeline, TimelineEventItemId};
 use ruma::{OwnedEventId, OwnedUserId};
-use tokio::{
-    sync::mpsc,
-    task::{JoinHandle, JoinSet},
-};
+use tokio::sync::mpsc;
 
+use super::executor::{EngineBoxFuture, JoinHandle, TaskSet, spawn};
 use crate::{
     Error, ErrorKind, Event, EventRef, HistoryState, MessageContent, MessageEditContent,
     MessageMentions, RichText, TimelineCommand, TimelineId,
@@ -44,10 +43,10 @@ impl Supervisor {
     pub(super) fn spawn(
         &self,
         label: &'static str,
-        task: impl Future<Output = ()> + Send + 'static,
+        task: impl Future<Output = ()> + SendOutsideWasm + 'static,
     ) -> JoinHandle<()> {
         let panics = self.0.clone();
-        tokio::spawn(async move {
+        spawn(async move {
             if AssertUnwindSafe(task).catch_unwind().await.is_err() {
                 let _ = panics.send(label);
             }
@@ -166,11 +165,11 @@ impl TimelineSession {
         &self,
         command: TimelineCommand,
         reply: Responder,
-        work: &mut JoinSet<()>,
+        work: &mut TaskSet<()>,
     ) {
         let timeline = self.timeline.clone();
         let thread_root = self.id.thread_root().cloned();
-        let background: Pin<Box<dyn Future<Output = Result<(), Error>> + Send>> = match command {
+        let background: EngineBoxFuture<'static, Result<(), Error>> = match command {
             TimelineCommand::Paginate { limit, .. } => return self.paginate(limit, Some(reply)),
             // network-bound, so it runs beside the actor instead of queueing the room behind it
             TimelineCommand::Redact { target, reason, .. } => Box::pin(async move {
@@ -331,7 +330,7 @@ fn item_id(target: EventRef) -> TimelineEventItemId {
 struct ReplyFetcher {
     timeline: Arc<Timeline>,
     asked: HashSet<OwnedEventId>,
-    fetches: JoinSet<()>,
+    fetches: TaskSet<()>,
 }
 
 impl ReplyFetcher {
@@ -369,7 +368,7 @@ async fn start_watcher(
     let mut replies = ReplyFetcher {
         timeline: timeline.clone(),
         asked: HashSet::new(),
-        fetches: JoinSet::new(),
+        fetches: TaskSet::new(),
     };
     replies.fetch(normalised.missing_replies);
     let id = id.clone();

Modified file: src/engine/watchers.rs+7 additions−11 deletions@@ -1,7 +1,4 @@
-use std::{
-    sync::Arc,
-    time::{Duration, Instant},
-};
+use std::{sync::Arc, time::Duration};
 
 use futures_util::StreamExt;
 use matrix_sdk::{
@@ -10,11 +7,10 @@ use matrix_sdk::{
     send_queue::{LocalEchoContent, RoomSendQueueUpdate},
 };
 use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
-use tokio::{
-    sync::{Notify, broadcast::error::RecvError, watch},
-    task::JoinSet,
-};
+use ruma::time::Instant;
+use tokio::sync::{Notify, broadcast::error::RecvError, watch};
 
+use super::executor::TaskSet;
 use crate::{
     ConnectionState, Event, encryption::encryption_state, error::classify_sdk_error,
     transport::EventSender,
@@ -106,7 +102,7 @@ pub(super) async fn connection(
         }
         if restart {
             sync.stop().await;
-            tokio::time::sleep(backoff).await;
+            matrix_sdk::sleep::sleep(backoff).await;
             backoff = (backoff * 2).min(SYNC_RESTART_MAX);
             sync.start().await;
         }
@@ -119,7 +115,7 @@ pub(super) async fn send_queue_errors(
     mut errors: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueRoomError>,
     online: watch::Receiver<ConnectionState>,
 ) {
-    let mut retries = JoinSet::new();
+    let mut retries = TaskSet::new();
     loop {
         let error = tokio::select! {
             error = errors.recv() => error,
@@ -130,7 +126,7 @@ pub(super) async fn send_queue_errors(
                 let client = client.clone();
                 let online = online.clone();
                 retries.spawn(async move {
-                    tokio::time::sleep(SEND_QUEUE_RETRY).await;
+                    matrix_sdk::sleep::sleep(SEND_QUEUE_RETRY).await;
                     if *online.borrow() == ConnectionState::Online
                         && let Some(room) = client.get_room(&error.room_id)
                     {

Modified file: src/error.rs+9 additions−3 deletions@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use matrix_sdk::{
-    HttpError, RefreshTokenError,
+    HttpError, RefreshTokenError, SendOutsideWasm, SyncOutsideWasm,
     authentication::oauth::error::{
         OAuthAuthorizationCodeError, OAuthDiscoveryError, OAuthError, OAuthTokenRevocationError,
         RequestTokenError,
@@ -17,6 +17,12 @@ use serde::{Deserialize, Serialize};
 
 use crate::InteractiveAuthChallenge;
 
+/// the sdk's `SendOutsideWasm` isn't an auto trait, so it can't go in a trait object
+#[cfg(not(target_family = "wasm"))]
+type SourceError = dyn std::error::Error + Send + Sync;
+#[cfg(target_family = "wasm")]
+type SourceError = dyn std::error::Error;
+
 /// the one thing that goes wrong, for commands and queries alike. match on `kind`, show `message`
 #[derive(Clone, Debug, thiserror::Error)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
@@ -29,7 +35,7 @@ pub struct Error {
     pub challenge: Option<Box<InteractiveAuthChallenge>>,
     #[source]
     #[cfg_attr(feature = "serde", serde(skip))]
-    source: Option<Arc<dyn std::error::Error + Send + Sync>>,
+    source: Option<Arc<SourceError>>,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -75,7 +81,7 @@ impl Error {
     pub(crate) fn with_source(
         kind: ErrorKind,
         message: impl Into<String>,
-        source: impl std::error::Error + Send + Sync + 'static,
+        source: impl std::error::Error + SendOutsideWasm + SyncOutsideWasm + 'static,
     ) -> Self {
         Self {
             source: Some(Arc::new(source)),

macltok

Commit 1fe480c16a3c

Commit `1fe480c16a3c`