fix(group): make MLS group thread safe #1349

bindings_ffi/src/mls.rs

@@ -1396,7 +1396,7 @@ impl FfiConversation {
 
     pub fn group_image_url_square(&self) -> Result<String, GenericError> {
         let provider = self.inner.mls_provider()?;
-        Ok(self.inner.group_image_url_square(provider)?)
+        Ok(self.inner.group_image_url_square(&provider)?)
     }
 
     pub async fn update_group_description(
@@ -1412,7 +1412,7 @@ impl FfiConversation {
 
     pub fn group_description(&self) -> Result<String, GenericError> {
         let provider = self.inner.mls_provider()?;
-        Ok(self.inner.group_description(provider)?)
+        Ok(self.inner.group_description(&provider)?)
     }
 
     pub async fn update_group_pinned_frame_url(
@@ -1546,7 +1546,9 @@ impl FfiConversation {
 
     pub fn group_metadata(&self) -> Result<Arc<FfiConversationMetadata>, GenericError> {
         let provider = self.inner.mls_provider()?;
-        let metadata = self.inner.metadata(provider)?;
+        let metadata = tokio::task::block_in_place(|| {
+            futures::executor::block_on(self.inner.metadata(&provider))
+        })?;
         Ok(Arc::new(FfiConversationMetadata {
             inner: Arc::new(metadata),
         }))
@@ -1558,7 +1560,9 @@ impl FfiConversation {
 
     pub fn conversation_type(&self) -> Result<FfiConversationType, GenericError> {
         let provider = self.inner.mls_provider()?;
-        let conversation_type = self.inner.conversation_type(&provider)?;
+        let conversation_type = tokio::task::block_in_place(|| {
+            futures::executor::block_on(self.inner.conversation_type(&provider))
+        })?;
         Ok(conversation_type.into())
     }
 }

bindings_node/src/conversation.rs

@@ -167,9 +167,10 @@ impl Conversation {
       self.created_at_ns,
     );
     let provider = group.mls_provider().map_err(ErrorWrapper::from)?;
-    let conversation_type = group
-      .conversation_type(&provider)
-      .map_err(ErrorWrapper::from)?;
+    let conversation_type = tokio::task::block_in_place(|| {
+      futures::executor::block_on(group.conversation_type(&provider))
+    })
+    .map_err(ErrorWrapper::from)?;
     let kind = match conversation_type {
       ConversationType::Group => None,
       ConversationType::Dm => Some(XmtpGroupMessageKind::Application),
@@ -248,7 +249,7 @@ impl Conversation {
     );
 
     let admin_list = group
-      .admin_list(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .admin_list(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(admin_list)
@@ -263,7 +264,7 @@ impl Conversation {
     );
 
     let super_admin_list = group
-      .super_admin_list(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .super_admin_list(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(super_admin_list)
@@ -449,7 +450,7 @@ impl Conversation {
     );
 
     let group_name = group
-      .group_name(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .group_name(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(group_name)
@@ -480,7 +481,7 @@ impl Conversation {
     );
 
     let group_image_url_square = group
-      .group_image_url_square(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .group_image_url_square(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(group_image_url_square)
@@ -511,7 +512,7 @@ impl Conversation {
     );
 
     let group_description = group
-      .group_description(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .group_description(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(group_description)
@@ -542,7 +543,7 @@ impl Conversation {
     );
 
     let group_pinned_frame_url = group
-      .group_pinned_frame_url(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .group_pinned_frame_url(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(group_pinned_frame_url)
@@ -585,7 +586,7 @@ impl Conversation {
 
     Ok(
       group
-        .is_active(group.mls_provider().map_err(ErrorWrapper::from)?)
+        .is_active(&group.mls_provider().map_err(ErrorWrapper::from)?)
         .map_err(ErrorWrapper::from)?,
     )
   }
@@ -609,9 +610,10 @@ impl Conversation {
       self.created_at_ns,
     );
 
-    let metadata = group
-      .metadata(group.mls_provider().map_err(ErrorWrapper::from)?)
-      .map_err(ErrorWrapper::from)?;
+    let metadata = tokio::task::block_in_place(|| {
+      futures::executor::block_on(group.metadata(&group.mls_provider()?))
+    })
+    .map_err(ErrorWrapper::from)?;
 
     Ok(GroupMetadata { inner: metadata })
   }

bindings_wasm/src/conversation.rs

@@ -180,14 +180,18 @@ impl Conversation {
   }
 
   #[wasm_bindgen(js_name = findMessages)]
-  pub fn find_messages(&self, opts: Option<ListMessagesOptions>) -> Result<Vec<Message>, JsError> {
+  pub async fn find_messages(
+    &self,
+    opts: Option<ListMessagesOptions>,
+  ) -> Result<Vec<Message>, JsError> {
     let opts = opts.unwrap_or_default();
     let group = self.to_mls_group();
     let provider = group
       .mls_provider()
       .map_err(|e| JsError::new(&format!("{e}")))?;
     let conversation_type = group
       .conversation_type(&provider)
+      .await
       .map_err(|e| JsError::new(&format!("{e}")))?;
     let kind = match conversation_type {
       ConversationType::Group => None,
@@ -238,7 +242,7 @@ impl Conversation {
     let group = self.to_mls_group();
     let admin_list = group
       .admin_list(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -252,7 +256,7 @@ impl Conversation {
     let group = self.to_mls_group();
     let super_admin_list = group
       .super_admin_list(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -398,7 +402,7 @@ impl Conversation {
 
     let group_name = group
       .group_name(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -428,7 +432,7 @@ impl Conversation {
 
     let group_image_url_square = group
       .group_image_url_square(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -455,7 +459,7 @@ impl Conversation {
 
     let group_description = group
       .group_description(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -485,7 +489,7 @@ impl Conversation {
 
     let group_pinned_frame_url = group
       .group_pinned_frame_url(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -505,7 +509,7 @@ impl Conversation {
 
     group
       .is_active(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -522,14 +526,15 @@ impl Conversation {
   }
 
   #[wasm_bindgen(js_name = groupMetadata)]
-  pub fn group_metadata(&self) -> Result<GroupMetadata, JsError> {
+  pub async fn group_metadata(&self) -> Result<GroupMetadata, JsError> {
     let group = self.to_mls_group();
     let metadata = group
       .metadata(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
+      .await
       .map_err(|e| JsError::new(&format!("{e}")))?;
 
     Ok(GroupMetadata { inner: metadata })

common/src/test.rs

@@ -38,7 +38,7 @@ pub fn logger() {
                 .from_env_lossy()
         };
 
-        tracing_subscriber::registry()
+        let _ = tracing_subscriber::registry()
             // structured JSON logger only if STRUCTURED=true
             .with(is_structured.then(|| {
                 tracing_subscriber::fmt::layer()
@@ -61,7 +61,7 @@ pub fn logger() {
                     })
                     .with_filter(filter())
             }))
-            .init();
+            .try_init();
     });
 }
 

examples/cli/serializable.rs

@@ -31,11 +31,12 @@ impl SerializableGroup {
 
         let metadata = group
             .metadata(
-                group
+                &group
                     .mls_provider()
                     .expect("MLS Provider could not be created"),
             )
-            .expect("could not load metadata");
+            .await
+            .unwrap();
         let permissions = group.permissions().expect("could not load permissions");
 
         Self {

xmtp_mls/src/client.rs

@@ -862,20 +862,22 @@ where
             .map(|group| {
                 let active_group_count = Arc::clone(&active_group_count);
                 async move {
-                    let mls_group = group.load_mls_group(provider)?;
                     tracing::info!(
                         inbox_id = self.inbox_id(),
                         "[{}] syncing group",
                         self.inbox_id()
                     );
                     tracing::info!(
                         inbox_id = self.inbox_id(),
-                        group_epoch = mls_group.epoch().as_u64(),
-                        "current epoch for [{}] in sync_all_groups() is Epoch: [{}]",
-                        self.inbox_id(),
-                        mls_group.epoch()
+                        "[{}] syncing group",
+                        self.inbox_id()
                     );
-                    if mls_group.is_active() {
+                    let is_active = group
+                        .load_mls_group_with_lock_async(provider, |mls_group| async move {
+                            Ok::<bool, GroupError>(mls_group.is_active())
+                        })
+                        .await?;
+                    if is_active {
                         group.maybe_update_installations(provider, None).await?;
 
                         group.sync_with_conn(provider).await?;

xmtp_mls/src/groups/intents.rs

@@ -865,9 +865,10 @@ pub(crate) mod tests {
         };
 
         let provider = group.client.mls_provider().unwrap();
-        let mut openmls_group = group.load_mls_group(&provider).unwrap();
-        let decrypted_message = openmls_group
-            .process_message(&provider, mls_message)
+        let decrypted_message = group
+            .load_mls_group_with_lock(&provider, |mut mls_group| {
+                Ok(mls_group.process_message(&provider, mls_message).unwrap())
+            })
             .unwrap();
 
         let staged_commit = match decrypted_message.into_content() {

xmtp_mls/src/groups/members.rs

@@ -40,9 +40,9 @@ where
         &self,
         provider: &XmtpOpenMlsProvider,
     ) -> Result<Vec<GroupMember>, GroupError> {
-        let openmls_group = self.load_mls_group(provider)?;
-        // TODO: Replace with try_into from extensions
-        let group_membership = extract_group_membership(openmls_group.extensions())?;
+        let group_membership = self.load_mls_group_with_lock(provider, |mls_group| {
+            Ok(extract_group_membership(mls_group.extensions())?)
+        })?;
         let requests = group_membership
             .members
             .into_iter()