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,7 @@ 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 = self.inner.metadata(&provider)?;
         Ok(Arc::new(FfiConversationMetadata {
             inner: Arc::new(metadata),
         }))

bindings_node/src/conversation.rs

@@ -248,7 +248,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 +263,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 +449,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 +480,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 +511,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 +542,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 +585,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)?,
     )
   }
@@ -610,7 +610,7 @@ impl Conversation {
     );
 
     let metadata = group
-      .metadata(group.mls_provider().map_err(ErrorWrapper::from)?)
+      .metadata(&group.mls_provider().map_err(ErrorWrapper::from)?)
       .map_err(ErrorWrapper::from)?;
 
     Ok(GroupMetadata { inner: metadata })

bindings_wasm/src/conversation.rs

@@ -180,7 +180,10 @@ 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
@@ -238,7 +241,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 +255,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 +401,7 @@ impl Conversation {
 
     let group_name = group
       .group_name(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -428,7 +431,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 +458,7 @@ impl Conversation {
 
     let group_description = group
       .group_description(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -485,7 +488,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 +508,7 @@ impl Conversation {
 
     group
       .is_active(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )
@@ -526,7 +529,7 @@ impl Conversation {
     let group = self.to_mls_group();
     let metadata = group
       .metadata(
-        group
+        &group
           .mls_provider()
           .map_err(|e| JsError::new(&format!("{e}")))?,
       )

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,7 +31,7 @@ impl SerializableGroup {
 
         let metadata = group
             .metadata(
-                group
+                &group
                     .mls_provider()
                     .expect("MLS Provider could not be created"),
             )

xmtp_mls/src/client.rs

@@ -873,20 +873,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()