mls write fix

xmtp_mls/Cargo.toml

@@ -24,7 +24,7 @@ bench = [
     "dep:fdlimit",
     "dep:ethers",
     "dep:const_format",
-    "xmtp_common/bench"
+    "xmtp_common/bench",
 ]
 default = ["grpc-api"]
 grpc-api = ["dep:xmtp_api_grpc"]
@@ -47,13 +47,15 @@ update-schema = ["toml"]
 aes-gcm = { version = "0.10.3", features = ["std"] }
 async-trait.workspace = true
 bincode.workspace = true
+bytes.workspace = true
 diesel_migrations.workspace = true
 futures = { workspace = true, features = ["alloc"] }
 hex.workspace = true
 hkdf.workspace = true
 openmls_rust_crypto = { workspace = true }
 openmls_traits = { workspace = true }
 parking_lot.workspace = true
+pin-project-lite.workspace = true
 prost = { workspace = true, features = ["prost-derive"] }
 rand = { workspace = true }
 reqwest = { workspace = true }
@@ -69,8 +71,6 @@ tracing.workspace = true
 trait-variant.workspace = true
 xmtp_common.workspace = true
 zeroize.workspace = true
-pin-project-lite.workspace = true
-bytes.workspace = true
 
 # XMTP/Local
 xmtp_content_types = { path = "../xmtp_content_types" }
@@ -108,13 +108,13 @@ diesel = { workspace = true, features = [
     "r2d2",
     "returning_clauses_for_sqlite_3_35",
     "sqlite",
-    "32-column-tables"
+    "32-column-tables",
 ] }
 dyn-clone.workspace = true
 libsqlite3-sys = { workspace = true }
 openmls.workspace = true
-openssl-sys.workspace = true
 openssl.workspace = true
+openssl-sys.workspace = true
 tokio = { workspace = true, features = [
     "macros",
     "tracing",

xmtp_mls/src/groups/mls_sync.rs

@@ -527,9 +527,21 @@ where
             ..
         } = *envelope;
 
-        provider.conn_ref().enable_readonly();
-        let processed_message = mls_group.process_message(provider, message)?;
-        provider.conn_ref().disable_readonly();
+        // We need to process the message twice to avoid an async transaction.
+        // We'll process for the first time, get the processed message,
+        // and roll the transaction back, so we can fetch updates from the server before
+        // being ready to process the message for a second time.
+        let mut processed_message = None;
+        let result = provider.transaction(|provider| {
+            processed_message = Some(mls_group.process_message(provider, message.clone()));
+            // Rollback the transaction. We want to synchronize with the server before committing.
+            Err::<(), StorageError>(StorageError::IntentionalRollback)
+        });
+        if !matches!(result, Err(StorageError::IntentionalRollback)) {
+            result?;
+        }
+
+        let processed_message = processed_message.expect("Was just set to Some")?;
 
         let (sender_inbox_id, sender_installation_id) =
             extract_message_sender(mls_group, &processed_message, envelope_timestamp_ns)?;
@@ -565,6 +577,8 @@ where
         };
 
         provider.transaction(|provider| {
+            let processed_message = mls_group.process_message(provider, message)?;
+
             if let Some(cursor) = cursor {
                 let is_updated = provider.conn_ref().update_cursor(
                     &envelope.group_id,

xmtp_mls/src/groups/validated_commit.rs

@@ -246,13 +246,12 @@ impl ValidatedCommit {
         let group_permissions: GroupMutablePermissions = extensions.try_into()?;
         let current_group_members = get_current_group_members(openmls_group);
 
-        let existing_group_extensions = openmls_group.extensions();
         let new_group_extensions = staged_commit.group_context().extensions();
 
         let metadata_changes = extract_metadata_changes(
             &immutable_metadata,
             &mutable_metadata,
-            existing_group_extensions,
+            extensions,
             new_group_extensions,
         )?;
 
@@ -296,7 +295,7 @@ impl ValidatedCommit {
             conn,
             &client,
             staged_commit,
-            existing_group_extensions,
+            extensions,
             &immutable_metadata,
             &mutable_metadata,
         )

xmtp_mls/src/storage/encrypted_store/wasm.rs

@@ -1,10 +1,9 @@
 //! WebAssembly specific connection for a SQLite Database
 //! Stores a single connection behind a mutex that's used for every libxmtp operation
-use std::sync::Arc;
-
 use diesel::{connection::AnsiTransactionManager, prelude::*};
 use parking_lot::Mutex;
 pub use sqlite_web::connection::WasmSqliteConnection as SqliteConnection;
+use std::sync::Arc;
 
 use super::{db_connection::DbConnectionPrivate, StorageError, StorageOption, XmtpDb};
 

xmtp_mls/src/storage/errors.rs

@@ -50,6 +50,8 @@ pub enum StorageError {
     Duplicate(DuplicateItem),
     #[error(transparent)]
     OpenMlsStorage(#[from] SqlKeyStoreError),
+    #[error("Transaction was intentionally rolled back")]
+    IntentionalRollback,
 }
 
 #[derive(Error, Debug)]