Fix Storage(NotFound("Intent $X ToPublish"))

xmtp_mls/src/groups/mod.rs

@@ -1460,15 +1460,15 @@ mod tests {
         // Check Amal's MLS group state.
         let amal_db = amal.context.store.conn().unwrap();
         let amal_mls_group = amal_group
-            .load_mls_group(&amal.mls_provider(amal_db.clone()))
+            .load_mls_group(amal.mls_provider(amal_db.clone()))
             .unwrap();
         let amal_members: Vec<Member> = amal_mls_group.members().collect();
         assert_eq!(amal_members.len(), 3);
 
         // Check Bola's MLS group state.
         let bola_db = bola.context.store.conn().unwrap();
         let bola_mls_group = bola_group
-            .load_mls_group(&bola.mls_provider(bola_db.clone()))
+            .load_mls_group(bola.mls_provider(bola_db.clone()))
             .unwrap();
         let bola_members: Vec<Member> = bola_mls_group.members().collect();
         assert_eq!(bola_members.len(), 3);

xmtp_mls/src/groups/sync.rs

@@ -1210,3 +1210,40 @@ fn get_removed_leaf_nodes(
         .map(|member| member.index)
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::builder::ClientBuilder;
+    use std::sync::Arc;
+    use xmtp_cryptography::utils::generate_local_wallet;
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn publish_intents_worst_case_scenario() {
+        let wallet = generate_local_wallet();
+        let amal = Arc::new(ClientBuilder::new_test_client(&wallet).await);
+        let amal_group: Arc<MlsGroup> =
+            Arc::new(amal.create_group(None, Default::default()).unwrap());
+
+        amal_group.send_message_optimistic(b"1").unwrap();
+        amal_group.send_message_optimistic(b"2").unwrap();
+        amal_group.send_message_optimistic(b"3").unwrap();
+        amal_group.send_message_optimistic(b"4").unwrap();
+        amal_group.send_message_optimistic(b"5").unwrap();
+        amal_group.send_message_optimistic(b"6").unwrap();
+
+        let mut futures = vec![];
+        let conn = amal.context().store.conn().unwrap();
+
+        for _ in 0..10 {
+            let client = amal.clone();
+            let conn = conn.clone();
+            let group = amal_group.clone();
+
+            futures.push(async move {
+                group.publish_intents(conn, &client).await.unwrap();
+            });
+        }
+        futures::future::join_all(futures).await;
+    }
+}

xmtp_mls/src/groups/validated_commit.rs

@@ -195,11 +195,11 @@ impl MetadataFieldChange {
  * Commit Validation Rules:
  * 1. If the `sequence_id` for an inbox has changed, it can only increase
  * 2. The client must create an expected diff of installations added and removed based on the difference between the current
- * [`GroupMembership`] and the [`GroupMembership`] found in the [`StagedCommit`]
+ *      [`GroupMembership`] and the [`GroupMembership`] found in the [`StagedCommit`]
  * 3. Installations may only be added or removed in the commit if they were added/removed in the expected diff
  * 4. For updates (either updating a path or via an Update Proposal) clients must verify that the `installation_id` is
- * present in the [`AssociationState`] for the `inbox_id` presented in the credential at the `to_sequence_id` found in the
- * new [`GroupMembership`].
+ *      present in the [`AssociationState`] for the `inbox_id` presented in the credential at the `to_sequence_id` found in the
+ *      new [`GroupMembership`].
  * 5. All proposals in a commit must come from the same installation
  * 6. No PSK proposals will be allowed
  */

xmtp_mls/src/storage/encrypted_store/group_intent.rs

@@ -195,7 +195,11 @@ impl DbConnection {
                 .filter(dsl::id.eq(intent_id))
                 // State machine requires that the only valid state transition to Published is from
                 // ToPublish
-                .filter(dsl::state.eq(IntentState::ToPublish))
+                .filter(
+                    dsl::state
+                        .eq(IntentState::ToPublish)
+                        .or(dsl::state.eq(IntentState::Published)),
+                )
                 .set((
                     dsl::state.eq(IntentState::Published),
                     dsl::payload_hash.eq(payload_hash),