dont err if intent was already published

xmtp_mls/src/groups/mls_sync.rs

@@ -1633,16 +1633,19 @@ pub(crate) mod tests {
         let amal_group_a: Arc<MlsGroup<_>> =
             Arc::new(amal_a.create_group(None, Default::default()).unwrap());
 
-        for _ in 0..500 {
+        let conn = amal_a.context().store().conn().unwrap();
+        let provider: Arc<XmtpOpenMlsProvider> = Arc::new(conn.into());
+
+        amal_group_a.sync().await.unwrap();
+        assert_eq!(provider.conn_ref().intents_deleted(), 1);
+
+        for _ in 0..100 {
             let s = xmtp_common::rand_string::<100>();
             amal_group_a.send_message_optimistic(s.as_bytes()).unwrap();
         }
 
-        let conn = amal_a.context().store().conn().unwrap();
-        let provider: Arc<XmtpOpenMlsProvider> = Arc::new(conn.into());
-
         let mut set = tokio::task::JoinSet::new();
-        for _ in 0..100 {
+        for _ in 0..50 {
             let g = amal_group_a.clone();
             let p = provider.clone();
             set.spawn(async move { g.publish_intents(&p).await });
@@ -1655,9 +1658,9 @@ pub(crate) mod tests {
         });
 
         let published = provider.conn_ref().intents_published();
-        assert_eq!(published, 501);
-        // let created = provider.conn_ref().intents_created();
-        // assert_eq!(created, 501);
+        assert_eq!(published, 101);
+        let created = provider.conn_ref().intents_created();
+        assert_eq!(created, 101);
         if !errs.is_empty() {
             panic!("Errors during publish");
         }

xmtp_mls/src/storage/encrypted_store/group_intent.rs

@@ -205,7 +205,7 @@ impl DbConnection {
                 .filter(dsl::state.eq(IntentState::ToPublish))
                 .set((
                     dsl::state.eq(IntentState::Published),
-                    dsl::payload_hash.eq(payload_hash),
+                    dsl::payload_hash.eq(&payload_hash),
                     dsl::post_commit_data.eq(post_commit_data),
                     dsl::staged_commit.eq(staged_commit),
                     dsl::published_in_epoch.eq(published_in_epoch),
@@ -214,10 +214,22 @@ impl DbConnection {
         })?;
 
         match res {
-            // If nothing matched the query, return an error. Either ID or state was wrong
-            0 => Err(StorageError::NotFound(format!(
-                "ToPublish intent {intent_id} for publish"
-            ))),
+            // If nothing matched the query, check if its already published, otherwise return an error. Either ID or state was wrong
+            0 => {
+                let already_published = self.raw_query(|conn| {
+                    dsl::group_intents
+                        .filter(dsl::id.eq(intent_id))
+                        .first::<StoredGroupIntent>(conn)
+                });
+
+                if already_published.is_ok() {
+                    return Ok(());
+                } else {
+                    return Err(StorageError::NotFound(format!(
+                        "Published intent {intent_id} for commit"
+                    )));
+                }
+            }
             _ => Ok(()),
         }
     }