xmtp_mls/groups/
oneshot.rs

1use super::{GroupError, MlsGroup, PreconfiguredPolicies};
2use crate::context::XmtpSharedContext;
3use xmtp_common::RetryableError;
4use xmtp_common::snippet::Snippet;
5use xmtp_db::{
6    MlsProviderExt,
7    TransactionOutcome::{self, Continue},
8    XmtpMlsStorageProvider,
9    consent_record::{ConsentState, conversation_consent},
10    group::ConversationType,
11    group_intent::IntentState,
12    prelude::{QueryGroup, QueryGroupIntent, QueryReaddStatus},
13};
14use xmtp_id::AsIdRef;
15use xmtp_mls_group_state::{group::GroupMetadataOptions, group_metadata::GroupMetadata};
16use xmtp_proto::{
17    types::{Cursor, GroupId},
18    xmtp::mls::message_contents::{OneshotMessage, oneshot_message},
19};
20
21pub struct Oneshot {}
22
23impl Oneshot {
24    /// Creates a oneshot group with the given message and adds the specified inbox IDs to it.
25    ///
26    /// A oneshot group is a special type of group that contains a single message and is used
27    /// for specific messaging scenarios like readd requests where no long-lived group is needed.
28    ///
29    /// # Arguments
30    /// * `context` - The shared context for group operations
31    /// * `inbox_ids` - List of inbox IDs to add to the group. Note that the sender's other
32    ///   installations are implicitly included and will also receive the oneshot message.
33    /// * `oneshot_message` - The oneshot message to include in the group metadata
34    ///
35    /// # Returns
36    /// An error if sending failed, otherwise nothing
37    pub async fn send_message<C: XmtpSharedContext, S: AsIdRef>(
38        context: C,
39        inbox_ids: impl AsRef<[S]>,
40        oneshot_message: OneshotMessage,
41    ) -> Result<(), GroupError> {
42        Self::send_message_with_outcome(context, inbox_ids, oneshot_message)
43            .await
44            .map_err(|(error, _)| error)
45    }
46
47    pub(super) async fn send_message_with_outcome<C: XmtpSharedContext, S: AsIdRef>(
48        context: C,
49        inbox_ids: impl AsRef<[S]>,
50        oneshot_message: OneshotMessage,
51    ) -> Result<(), (GroupError, bool)> {
52        // Create a oneshot group with the oneshot message
53        let group = MlsGroup::<C>::create_and_insert(
54            context.clone(),
55            ConversationType::Oneshot,
56            PreconfiguredPolicies::default().to_policy_set(),
57            GroupMetadataOptions::default(),
58            Some(oneshot_message),
59        )
60        .map_err(|error| (error, true))?;
61
62        let sent = if !inbox_ids.as_ref().is_empty() {
63            group.add_members(&inbox_ids).await.map(|_| ())
64        } else {
65            group.add_missing_installations().await.map(|_| ())
66        };
67        // A delayed follow-up can carry a rate limit after publication.
68        let refused = sent.as_ref().err().is_some_and(|error| {
69            !matches!(error, GroupError::PublishedButUnconfirmed { .. })
70                && error.rate_limit_delay().is_some()
71        });
72        let cleanup = delete_oneshot_if_settled(&group, refused);
73        match sent {
74            Err(error) => Err((error, refused || cleanup.unwrap_or(false))),
75            Ok(()) => cleanup.map(|_| ()).map_err(|error| (error, false)),
76        }
77    }
78
79    pub fn process_message(
80        provider: &impl MlsProviderExt,
81        own_inbox_id: &str,
82        sender_installation_id: Vec<u8>,
83        message: OneshotMessage,
84    ) -> Result<(), GroupError> {
85        match message.message_type {
86            Some(oneshot_message::MessageType::ReaddRequest(readd_request)) => {
87                if Self::validate_readd_request(
88                    provider.key_store(),
89                    own_inbox_id,
90                    &readd_request.group_id,
91                    &sender_installation_id,
92                )? {
93                    let group_id = GroupId::try_from(readd_request.group_id.as_slice())?;
94                    provider.key_store().db().update_requested_at_sequence_id(
95                        &group_id,
96                        &sender_installation_id,
97                        readd_request.latest_commit_sequence_id as i64,
98                    )?;
99                    tracing::info!(
100                        group_id = readd_request.group_id.snippet(),
101                        sender_installation_id = sender_installation_id.snippet(),
102                        latest_commit_sequence_id = readd_request.latest_commit_sequence_id,
103                        "Stored incoming readd request"
104                    );
105                }
106            }
107            _ => {
108                tracing::warn!(
109                    "Oneshot message {:?} is not a recognized message type",
110                    message.message_type
111                );
112            }
113        }
114        Ok(())
115    }
116
117    pub fn process_welcome(
118        provider: &impl MlsProviderExt,
119        own_inbox_id: &str,
120        cursor: Cursor,
121        sender_installation_id: Vec<u8>,
122        metadata: GroupMetadata,
123    ) -> Result<(), GroupError> {
124        tracing::debug!("Processing oneshot welcome");
125        if let Some(message) = metadata.oneshot_message {
126            Self::process_message(provider, own_inbox_id, sender_installation_id, message)?;
127        } else {
128            tracing::warn!(
129                "Oneshot group welcome {} does not have oneshot message",
130                cursor
131            );
132        }
133        Ok(())
134    }
135
136    // implements: FORK-053
137    fn validate_readd_request(
138        storage: &impl XmtpMlsStorageProvider,
139        own_inbox_id: &str,
140        group_id: &Vec<u8>,
141        sender_installation_id: &Vec<u8>,
142    ) -> Result<bool, GroupError> {
143        let id = GroupId::try_from(group_id.as_slice())?;
144        if conversation_consent(&storage.db(), own_inbox_id, &id)? != ConsentState::Allowed {
145            tracing::warn!(
146                group_id = group_id.snippet(),
147                "Group not consented for readd request"
148            );
149            return Ok(false);
150        }
151        // Fetch the OpenMLS group without locking; consistency is not important here, and we are not writing to it
152        let Ok(Some(openmls_group)) = openmls::group::MlsGroup::load(
153            storage,
154            &openmls::group::GroupId::from_slice(group_id.as_slice()),
155        ) else {
156            tracing::warn!(
157                group_id = group_id.snippet(),
158                "Unable to load OpenMLS group for readd request"
159            );
160            return Ok(false);
161        };
162        if !openmls_group
163            .members()
164            .any(|member| member.signature_key == *sender_installation_id)
165        {
166            tracing::warn!(
167                group_id = group_id.snippet(),
168                "Sender not a member of group for readd request"
169            );
170            return Ok(false);
171        }
172
173        Ok(true)
174    }
175}
176
177// Returns whether a failed send permits a fresh request.
178fn delete_oneshot_if_settled<C: XmtpSharedContext>(
179    group: &MlsGroup<C>,
180    refused: bool,
181) -> Result<bool, GroupError> {
182    if group.conversation_type != ConversationType::Oneshot {
183        return Ok(false);
184    }
185    let intents = group
186        .context
187        .db()
188        .find_group_intents(group.group_id, None, None)?;
189    let unsent = intents.iter().all(|intent| {
190        matches!(
191            intent.state,
192            IntentState::ToPublish | IntentState::Error | IntentState::Superseded
193        )
194    });
195    if !refused
196        && !unsent
197        && intents.iter().any(|intent| {
198            matches!(
199                intent.state,
200                IntentState::ToPublish | IntentState::Published | IntentState::Committed
201            )
202        })
203    {
204        return Ok(false);
205    }
206    crate::state_tx::state_write(group.context.mls_storage(), |tx| {
207        let storage = tx.storage();
208        let mut mls = openmls::group::MlsGroup::load(&storage, &group.group_id.to_openmls())?
209            .ok_or(xmtp_db::StorageError::from(xmtp_db::NotFound::MlsGroup(
210                group.group_id,
211            )))?;
212        mls.delete(&storage)?;
213        storage.db().delete_oneshot_group(&group.group_id)?;
214        Ok(Continue(unsent || refused))
215    })
216    .map(TransactionOutcome::into_continued)
217}
218
219#[cfg(all(test, not(target_arch = "wasm32")))]
220#[allow(dead_code, clippy::duplicate_mod, clippy::unwrap_used)]
221#[path = "mls_sync/publish/tests/faults.rs"]
222mod send_faults;
223#[cfg(all(test, not(target_arch = "wasm32")))]
224use xmtp_proto::backend_v1::client_envelope::Payload;
225
226#[cfg(all(test, not(target_arch = "wasm32")))]
227mod tests {
228    use super::send_faults::{
229        Fault, FaultyApi, FaultyContext, faulty_group, group_message, welcome,
230    };
231    use super::*;
232    use crate::groups::commit_log::CommitLogWorker;
233    use crate::tester;
234    use futures::stream::StreamExt;
235    use xmtp_common::{
236        RetryableError,
237        time::{Duration, sleep},
238    };
239    use xmtp_db::{ConnectionExt, diesel::prelude::*, prelude::*};
240
241    #[derive(QueryableByName, Debug, PartialEq)]
242    struct Count {
243        #[diesel(sql_type = diesel::sql_types::BigInt)]
244        n: i64,
245    }
246    fn send_state(context: &impl XmtpSharedContext) -> Vec<Count> {
247        context.db().raw_query(|conn| diesel::sql_query(
248            "SELECT COUNT(*) AS n FROM groups UNION ALL SELECT COUNT(*) FROM group_intents UNION ALL SELECT COUNT(*) FROM group_messages UNION ALL SELECT COUNT(*) FROM local_commit_log UNION ALL SELECT COUNT(*) FROM remote_commit_log UNION ALL SELECT COUNT(*) FROM refresh_state UNION ALL SELECT COUNT(*) FROM incoming_envelopes UNION ALL SELECT COUNT(*) FROM openmls_key_value"
249        ).load(conn)).expect("sender state counts")
250    }
251    const DELAY: Duration = Duration::from_millis(10);
252
253    fn request_message(group_id: GroupId) -> OneshotMessage {
254        OneshotMessage {
255            message_type: Some(MessageType::ReaddRequest(ReaddRequest {
256                group_id: group_id.to_vec(),
257                latest_commit_sequence_id: 42,
258            })),
259        }
260    }
261
262    async fn readd_pair(published: bool, fail: bool) -> (MlsGroup<FaultyContext>, FaultyApi) {
263        tester!(alix, disable_workers, enable_fork_recovery_requests);
264        tester!(bo, disable_workers);
265        let group = alix
266            .find_or_create_dm(bo.inbox_id().to_string(), None)
267            .await
268            .expect("requester fixture");
269        let mut worker = CommitLogWorker::new(alix.context.clone());
270        worker._tick().await.expect("requester fixture");
271        worker._tick().await.expect("requester fixture");
272        let faults = if fail {
273            vec![Fault::RateLimit(DELAY); 18]
274        } else {
275            vec![]
276        };
277        let target = if published { welcome } else { group_message };
278        let api = FaultyApi::new(&alix, target, faults);
279        let faulty = faulty_group(&alix, &group.group_id, api.clone()).await;
280        let db = faulty.context.db();
281        db.set_group_commit_log_forked_status(&group.group_id, Some(true))
282            .expect("requester fixture");
283        (faulty, api)
284    }
285    use xmtp_proto::xmtp::mls::message_contents::{ReaddRequest, oneshot_message::MessageType};
286
287    // verifies: FORK-051, SEND-007, SEND-015
288    #[rstest::rstest]
289    #[case::success(false, false, false)]
290    #[case::intent_insert(false, false, true)]
291    #[case::refused(false, true, false)]
292    #[case::published_but_unconfirmed(true, true, false)]
293    #[xmtp_common::test(unwrap_try = true)]
294    async fn oneshot_readd_send_outcomes(
295        #[case] published: bool,
296        #[case] fail: bool,
297        #[case] trigger: bool,
298    ) -> Result<(), xmtp_common::BoxDynError> {
299        let (group, api) = readd_pair(published, fail).await;
300        let context = &group.context;
301        let db = context.db();
302        let before = send_state(context);
303        if trigger {
304            db.raw_query(|conn| diesel::sql_query("CREATE TRIGGER fail_intent BEFORE INSERT ON group_intents BEGIN SELECT RAISE(ABORT, 'request not recorded before sending') WHERE NOT EXISTS (SELECT 1 FROM readd_status WHERE requested_at_sequence_id IS NOT NULL); SELECT RAISE(ABORT, 'injected intent failure'); END").execute(conn))?;
305        }
306        let mut worker = CommitLogWorker::new(group.context.clone());
307        let result = worker._tick().await;
308        assert_eq!(result.is_err(), fail || trigger);
309        if trigger {
310            assert!(format!("{result:?}").contains("injected intent failure"));
311        }
312        if fail {
313            assert_eq!(result.unwrap_err().rate_limit_delay(), Some(DELAY));
314        }
315        let awaiting = || {
316            db.is_awaiting_readd(&group.group_id, context.installation_id().as_slice())
317                .unwrap()
318        };
319        let keep = published || !fail && !trigger;
320        assert_eq!(awaiting(), keep, "requester latch");
321        let sent = api.requests.lock().len();
322        if published {
323            use xmtp_db::schema::groups;
324            let id = db.raw_query(|conn| {
325                groups::table
326                    .filter(groups::conversation_type.eq(ConversationType::Oneshot))
327                    .select(groups::id)
328                    .first::<GroupId>(conn)
329            })?;
330            let intent = db.find_group_intents(id, None, None)?.pop().unwrap();
331            assert_eq!(intent.state, IntentState::Committed);
332            let bytes = db.prepared_envelopes(intent.id)?.unwrap();
333            worker._tick().await?;
334            assert_eq!(api.requests.lock().len(), sent);
335            assert_eq!(db.prepared_envelopes(intent.id)?.unwrap(), bytes);
336            let temporary = MlsGroup::new_cached(context.clone(), &id)?.0;
337            temporary.sync().await?;
338            // A settled send can still have later queued input.
339            db.raw_query(|conn| diesel::sql_query("INSERT INTO incoming_envelopes (entity_id, entity_kind, sequence_id, envelope, blocked, retry_at_ns, envelope_bytes) VALUES (?, 2, 999999999, X'01', 1, 9223372036854775807, 1)").bind::<diesel::sql_types::Binary, _>(id.as_slice()).execute(conn))?;
340            assert!(delete_oneshot_if_settled(&temporary, false).is_ok());
341            assert_eq!(send_state(context), before);
342        } else {
343            assert_eq!(send_state(context), before);
344            if trigger {
345                db.raw_query(|conn| diesel::sql_query("DROP TRIGGER fail_intent").execute(conn))?;
346            }
347            sleep(DELAY).await;
348            worker._tick().await?;
349            assert_eq!(api.requests.lock().len() > sent, fail || trigger);
350        }
351        Ok(())
352    }
353
354    // verifies: FORK-052
355    #[xmtp_common::test(unwrap_try = true)]
356    async fn test_receive_oneshot_message_via_syncing() {
357        tester!(alix, disable_workers);
358        tester!(bo, disable_workers);
359        tester!(caro, disable_workers);
360        tester!(phone, from: alix, disable_workers);
361        let a_group = alix
362            .create_group_with_members(&[bo.inbox_id(), caro.inbox_id()], None, None)
363            .await?;
364        let group_id = a_group.group_id;
365        let own = alix.context.installation_id();
366        for recipient in [&bo, &caro] {
367            recipient.sync_all_welcomes_and_groups(None).await?;
368            recipient
369                .group(&group_id)?
370                .update_consent_state(ConsentState::Allowed)?;
371            let db = recipient.context.db();
372            assert!(db.get_readd_status(&group_id, own.as_slice())?.is_none());
373        }
374        phone.sync_all_welcomes_and_groups(None).await?;
375        phone
376            .group(&group_id)?
377            .update_consent_state(ConsentState::Allowed)?;
378        let before = send_state(&alix.context);
379        let message = request_message(group_id);
380        Oneshot::send_message(alix.context.clone(), Vec::<String>::new(), message).await?;
381        assert_eq!(send_state(&alix.context), before);
382        phone.sync_welcomes().await?;
383        let status = phone
384            .context
385            .db()
386            .get_readd_status(&group_id, own.as_slice())?
387            .unwrap();
388        assert_eq!(status.requested_at_sequence_id, Some(42));
389        assert_eq!(status.responded_at_sequence_id, None);
390        Oneshot::send_message(
391            alix.context.clone(),
392            [bo.inbox_id(), caro.inbox_id()],
393            request_message(group_id),
394        )
395        .await?;
396        assert_eq!(send_state(&alix.context), before);
397        for recipient in [&bo, &caro] {
398            recipient.sync_welcomes().await?;
399            assert_eq!(recipient.find_groups(Default::default())?.len(), 1);
400            let db = recipient.context.db();
401            let status = db.get_readd_status(&group_id, own.as_slice())?.unwrap();
402            assert_eq!(status.requested_at_sequence_id, Some(42));
403            assert_eq!(status.responded_at_sequence_id, None);
404        }
405    }
406
407    // verifies: FORK-052
408    #[xmtp_common::test(unwrap_try = true)]
409    async fn test_oneshot_groups_not_in_find_groups() {
410        tester!(alix);
411
412        // Keep a stored one-shot: sender cleanup must not bypass the listing filter.
413        MlsGroup::create_and_insert(
414            alix.context.clone(),
415            ConversationType::Oneshot,
416            PreconfiguredPolicies::default().to_policy_set(),
417            GroupMetadataOptions::default(),
418            None,
419        )?;
420        let normal = alix.create_group(None, None)?;
421        let alix_groups = alix.find_groups(Default::default()).unwrap();
422
423        // Oneshot groups should not appear in the regular groups list
424        assert_eq!(alix_groups.len(), 1);
425        assert_eq!(alix_groups[0].group_id, normal.group_id);
426    }
427
428    #[tokio::test]
429    async fn test_oneshot_groups_not_in_stream_groups() {
430        tester!(alix);
431        tester!(bo);
432
433        // Subscribe to conversation events
434        let mut alix_conversations = alix
435            .stream_conversations(None, false)
436            .await
437            .expect("Failed to stream conversations");
438        let mut bo_conversations = bo
439            .stream_conversations(None, false)
440            .await
441            .expect("Failed to stream conversations");
442
443        // Create a test oneshot message
444        let readd_request = ReaddRequest {
445            group_id: vec![5, 6, 7, 8],
446            latest_commit_sequence_id: 0,
447        };
448        let oneshot_message = OneshotMessage {
449            message_type: Some(MessageType::ReaddRequest(readd_request)),
450        };
451
452        // Alix sends the oneshot message to Bo
453        Oneshot::send_message(alix.context.clone(), vec![bo.inbox_id()], oneshot_message)
454            .await
455            .expect("Failed to send oneshot message");
456
457        // Small delay to ensure any events would have been processed
458        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
459
460        // Try to get any conversation events with a timeout
461        let alix_event = tokio::time::timeout(
462            std::time::Duration::from_millis(100),
463            alix_conversations.next(),
464        )
465        .await;
466        let bo_event = tokio::time::timeout(
467            std::time::Duration::from_millis(100),
468            bo_conversations.next(),
469        )
470        .await;
471
472        // Both should timeout (no new conversations)
473        assert!(
474            alix_event.is_err(),
475            "Alix should not receive new conversation event"
476        );
477        assert!(
478            bo_event.is_err(),
479            "Bo should not receive new conversation event"
480        );
481    }
482
483    #[tokio::test]
484    async fn test_syncing_and_streaming_oneshot_group_simultaneously() {
485        // Test syncing and streaming simultaneously, which causes Welcome to be processed twice
486        // Note that on the second time, there is no group in the DB to refetch by ID - this should
487        // not surface an error in the stream
488        tester!(alix);
489        tester!(bo);
490
491        let mut bo_conversations = bo
492            .stream_conversations(None, false)
493            .await
494            .expect("Failed to stream conversations");
495
496        // Create a test oneshot message
497        let readd_request = ReaddRequest {
498            group_id: vec![5, 6, 7, 8],
499            latest_commit_sequence_id: 0,
500        };
501        let oneshot_message = OneshotMessage {
502            message_type: Some(MessageType::ReaddRequest(readd_request)),
503        };
504
505        // Alix sends the oneshot message to Bo
506        Oneshot::send_message(alix.context.clone(), vec![bo.inbox_id()], oneshot_message)
507            .await
508            .expect("Failed to send oneshot message");
509
510        // Bo syncs welcomes to receive any oneshot groups
511        bo.sync_welcomes().await.expect("Failed to sync welcomes");
512
513        // Small delay to ensure any events would have been processed
514        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
515
516        let bo_event = tokio::time::timeout(
517            std::time::Duration::from_millis(100),
518            bo_conversations.next(),
519        )
520        .await;
521
522        assert!(
523            bo_event.is_err(),
524            "Bo should not receive new conversation event"
525        );
526    }
527}
xmtp_mls/groups/oneshot.rs

xmtp_mls/groups/
oneshot.rs
