xmtp_mls/groups/mls_sync/
receive.rs

1//! Receiving envelopes and storing transcript messages.
2
3use super::*;
4use crate::messages::decoded_message::DecodedMessage;
5use xmtp_db::group_message::SortBy;
6use xmtp_db::prelude::QueryGroupMessage;
7
8/// Rows read per page when scanning a DM's group updates for duplicates.
9const PAGE_SIZE: i64 = 100;
10
11impl<Context> MlsGroup<Context>
12where
13    Context: XmtpSharedContext,
14{
15    /// Wait for a fixed network prefix. The summary is local history, not proof of completion.
16    #[tracing::instrument(skip_all, level = "trace")]
17    pub async fn receive(&self) -> Result<ProcessSummary, GroupError> {
18        use xmtp_db::delivery::{DeliveryScope, QueryDelivery};
19        let db = self.context.db();
20        if db
21            .find_group(&self.group_id)?
22            .is_some_and(|group| group.membership_state == GroupMembershipState::Restored)
23        {
24            return Err(GroupError::GroupInactive);
25        }
26        let mut position = db.current_delivery_cursor()?;
27        crate::subscriptions::barrier::receive_through_current(
28            &self.context,
29            vec![xmtp_proto::types::Topic::new_group_message(self.group_id)],
30        )
31        .await?;
32        let upper = db.current_delivery_cursor()?;
33        let settings = self.context.incoming_runtime().policy();
34        let mut summary = ProcessSummary::default();
35        loop {
36            let rows = db.replay_delivery_messages_bounded(
37                position,
38                &DeliveryScope::Groups(vec![self.group_id]),
39                xmtp_common::time::now_ns(),
40                settings.max_local_read_rows,
41                settings.max_local_read_bytes,
42            )?;
43            if rows.is_empty() {
44                break;
45            }
46            let mut reached_upper = false;
47            for row in rows {
48                if row.cursor.delivery_sequence > upper.delivery_sequence {
49                    reached_upper = true;
50                    break;
51                }
52                position = row.cursor;
53                let message = row.message;
54                summary.add_id(message.cursor());
55                summary.add(MessageIdentifier {
56                    cursor: message.cursor(),
57                    group_id: message.group_id,
58                    created_ns: chrono::DateTime::from_timestamp_nanos(message.sent_at_ns),
59                    previously_processed: false,
60                    internal_id: Some(message.id),
61                    group_context: None,
62                    intent_kind: None,
63                });
64            }
65            if reached_upper || position.delivery_sequence >= upper.delivery_sequence {
66                break;
67            }
68        }
69        Ok(summary)
70    }
71
72    #[tracing::instrument(skip_all, level = "trace")]
73    pub(super) fn maybe_update_cursor(
74        &self,
75        db: &impl DbQuery,
76        message: &xmtp_proto::types::GroupMessage,
77    ) -> Result<bool, StorageError> {
78        let updated = db.update_cursor(
79            message.group_id,
80            xmtp_db::refresh_state::EntityKind::ApplicationMessage,
81            message.cursor,
82        )?;
83        if updated {
84            log_event!(
85                Event::GroupCursorUpdate,
86                self.context.installation_id(),
87                group_id = message.group_id.as_slice(),
88                cursor = message.cursor.0,
89            );
90        } else {
91            tracing::debug!("no cursor update required");
92        }
93        Ok(updated)
94    }
95
96    // implements: GMOD-034
97    pub(super) fn save_transcript_message(
98        &self,
99        validated_commit: ValidatedCommit,
100        timestamp_ns: u64,
101        cursor: Cursor,
102        storage: &impl XmtpMlsStorageProvider,
103    ) -> Result<Option<(StoredGroupMessage, GroupUpdated)>, GroupMessageProcessingError> {
104        if validated_commit.is_empty() {
105            return Ok(None);
106        }
107        let sender_installation_id = validated_commit.actor_installation_id();
108        let sender_inbox_id = validated_commit.actor_inbox_id();
109
110        let pending_remove_users = &storage.db().get_pending_remove_users(&self.group_id)?;
111        let payload: GroupUpdated = validated_commit.into_with(pending_remove_users);
112        tracing::info!("Storing transcript message");
113        let encoded_payload = GroupUpdatedCodec::encode(payload.clone())?;
114        let mut encoded_payload_bytes = Vec::new();
115        encoded_payload.encode(&mut encoded_payload_bytes)?;
116
117        let message_id = calculate_message_id(
118            self.group_id,
119            encoded_payload_bytes.as_slice(),
120            &timestamp_ns.to_string(),
121        );
122        let content_type = encoded_payload.r#type.unwrap_or_else(|| {
123            tracing::warn!("Missing content type in encoded payload, using default values");
124            // Default content type values
125            xmtp_proto::xmtp::mls::message_contents::ContentTypeId {
126                authority_id: "unknown".to_string(),
127                type_id: "unknown".to_string(),
128                version_major: 0,
129                version_minor: 0,
130            }
131        });
132
133        self.handle_metadata_update_from_commit(&payload.metadata_field_changes, storage)?;
134
135        // When a DM is stitched, it can repeat group updates. We want to prevent saving those messages.
136        if self.update_already_exists(&payload, storage)? {
137            return Ok(None);
138        }
139
140        let msg = StoredGroupMessage {
141            id: message_id,
142            group_id: self.group_id,
143            decrypted_message_bytes: encoded_payload_bytes,
144            sent_at_ns: timestamp_ns as i64,
145            kind: GroupMessageKind::MembershipChange,
146            sender_installation_id,
147            sender_inbox_id,
148            delivery_status: DeliveryStatus::Published,
149            content_type: ContentType::from_identifier(
150                &content_type.authority_id,
151                &content_type.type_id,
152                content_type.version_major,
153            ),
154            version_major: content_type.version_major as i32,
155            version_minor: content_type.version_minor as i32,
156            authority_id: content_type.authority_id.to_string(),
157            reference_id: None,
158            sequence_id: cursor.0 as i64,
159            envelope_hash: None,
160            expiry_ns: None,
161            expiry_unknown: false,
162            deleted_by_inbox_id: None,
163            expire_at_ns: None,
164            inserted_at_ns: 0, // Will be set by database
165            should_push: true,
166            // Matches the key used to derive `message_id` above.
167            idempotency_key: timestamp_ns.to_string(),
168        };
169
170        msg.store_or_ignore(&storage.db())?;
171        Ok(Some((msg, payload)))
172    }
173
174    fn update_already_exists(
175        &self,
176        payload: &GroupUpdated,
177        storage: &impl XmtpMlsStorageProvider,
178    ) -> Result<bool, GroupMessageProcessingError> {
179        if self.dm_id.is_none() || payload.added_inboxes.is_empty() {
180            // Only dedupe for DMs.
181            // Only dedupe for group adds.
182            return Ok(false);
183        }
184
185        let mut deduper = GroupUpdateDeduper::default();
186        let mut inserted_after_ns = None;
187        // Grows while a page holds a single `inserted_at_ns`, so that a tie
188        // group larger than one page still fits and the cursor can advance.
189        let mut limit = PAGE_SIZE;
190        loop {
191            // DMs are stitched, so we don't want to have the same
192            // group updates from multiple DMs being saved to the database.
193            //
194            // Sort by the same column the cursor advances on. The default sort
195            // is `SentAt`, which does not agree with an `inserted_at_ns`
196            // cursor: the last row by sent time need not hold the largest
197            // `inserted_at_ns`, so the cursor could stall and repeat a page
198            // forever.
199            //
200            // Read stored rows, not decoded messages. The page count must be
201            // the number of rows the database returned: a decoded view drops
202            // rows it cannot decode, which makes a full page look short and
203            // lets the rows past its edge be skipped. Decoding one row at a
204            // time also skips the reaction and reply lookups that a decoded
205            // list runs for every page, which group updates never need.
206            let rows = storage.db().get_group_messages(
207                &self.group_id,
208                &MsgQueryArgs {
209                    content_types: Some(vec![ContentType::GroupUpdated]),
210                    inserted_after_ns,
211                    limit: Some(limit),
212                    sort_by: Some(SortBy::InsertedAt),
213                    ..Default::default()
214                },
215            )?;
216
217            let full_page = rows.len() >= limit as usize;
218            let last_value = rows.last().map(|row| row.inserted_at_ns);
219
220            // The cursor filter is a strict `>` on `inserted_at_ns`, which the
221            // database records at millisecond granularity, so rows written in
222            // the same millisecond share a value. On a full page the final
223            // value may continue past the page edge, so hold those rows back
224            // and let the next page re-read them in full. A short page reached
225            // the end of the scan and keeps every row.
226            let cutoff = last_value.filter(|_| full_page);
227            let consumed = rows
228                .into_iter()
229                .filter(|row| Some(row.inserted_at_ns) != cutoff);
230
231            let mut highest_consumed = None;
232            for row in consumed {
233                highest_consumed = Some(row.inserted_at_ns);
234                let MessageBody::GroupUpdated(update) = DecodedMessage::from(row).content else {
235                    continue;
236                };
237
238                deduper.consume(&update);
239            }
240
241            match highest_consumed {
242                // Resume just above the last value read in full.
243                Some(value) => {
244                    inserted_after_ns = Some(value);
245                    limit = PAGE_SIZE;
246                }
247                // A full page of one repeated value: nothing could be retired,
248                // so re-read it larger rather than stall.
249                None if full_page => limit = limit.saturating_mul(2),
250                None => break,
251            }
252        }
253
254        Ok(deduper.is_dupe(payload))
255    }
256}
257
258#[cfg(test)]
259mod tests {
260    use super::*;
261    use crate::prelude::*;
262    use crate::tester;
263    use xmtp_common::{rand_vec, time::now_ns};
264    use xmtp_content_types::{ContentCodec, encryption::sha256, group_updated::GroupUpdatedCodec};
265    use xmtp_db::diesel::{ExpressionMethods, QueryDsl, RunQueryDsl};
266    use xmtp_db::group_message::{DeliveryStatus, GroupMessageKind};
267    use xmtp_db::{ConnectionExt, DbConnection, Store};
268    use xmtp_proto::types::GroupId;
269    use xmtp_proto::xmtp::mls::message_contents::group_updated::Inbox;
270
271    /// Marks the rows a test seeds, so they can be pinned to one `inserted_at_ns`.
272    const SEEDED_SENDER: &str = "seeded";
273
274    fn add_update(inbox_id: &str) -> GroupUpdated {
275        GroupUpdated {
276            added_inboxes: vec![Inbox {
277                inbox_id: inbox_id.to_string(),
278            }],
279            ..Default::default()
280        }
281    }
282
283    fn encode(update: &GroupUpdated) -> Vec<u8> {
284        let mut bytes = Vec::new();
285        GroupUpdatedCodec::encode(update.clone())
286            .unwrap()
287            .encode(&mut bytes)
288            .unwrap();
289        bytes
290    }
291
292    /// Store `rows` as group updates of `group_id`, in order, then pin them to
293    /// one `inserted_at_ns` one millisecond above every row the group already
294    /// holds.
295    fn seed_tied_updates<C: ConnectionExt>(
296        db: &DbConnection<C>,
297        group_id: GroupId,
298        rows: Vec<Vec<u8>>,
299    ) {
300        for (i, bytes) in rows.into_iter().enumerate() {
301            StoredGroupMessage {
302                id: sha256(&rand_vec::<12>()),
303                group_id,
304                decrypted_message_bytes: bytes,
305                sent_at_ns: now_ns(),
306                kind: GroupMessageKind::MembershipChange,
307                sender_installation_id: vec![1, 2, 3],
308                sender_inbox_id: SEEDED_SENDER.to_string(),
309                delivery_status: DeliveryStatus::Published,
310                content_type: ContentType::GroupUpdated,
311                version_major: 0,
312                version_minor: 0,
313                authority_id: "unknown".to_string(),
314                reference_id: None,
315                sequence_id: i as i64 + 1,
316                envelope_hash: None,
317                expiry_ns: None,
318                expiry_unknown: false,
319                deleted_by_inbox_id: None,
320                expire_at_ns: None,
321                inserted_at_ns: 0,
322                should_push: true,
323                idempotency_key: String::new(),
324            }
325            .store(db)
326            .unwrap();
327        }
328
329        // The database stamps `inserted_at_ns` at insert time, and its clock
330        // does not agree with `now_ns`. Pin the seeded rows to one value just
331        // above every row the group already holds, rather than rely on how
332        // fast the loop ran or on the wall clock.
333        db.raw_query(|conn| {
334            use xmtp_db::schema::group_messages::dsl;
335            let newest: Option<i64> = dsl::group_messages
336                .filter(dsl::group_id.eq(group_id))
337                .select(xmtp_db::diesel::dsl::max(dsl::inserted_at_ns))
338                .first(conn)?;
339            let shared_inserted_at_ns = newest.unwrap_or_default() + 1_000_000;
340            xmtp_db::diesel::update(dsl::group_messages)
341                .filter(dsl::group_id.eq(group_id))
342                .filter(dsl::sender_inbox_id.eq(SEEDED_SENDER))
343                .set(dsl::inserted_at_ns.eq(shared_inserted_at_ns))
344                .execute(conn)
345        })
346        .unwrap();
347    }
348
349    #[xmtp_common::test(unwrap_try = true)]
350    async fn test_startup_preserves_dm_update_history() {
351        use xmtp_db::user_preferences::StoredUserPreferences;
352        use xmtp_proto::xmtp::mls::message_contents::group_updated::MetadataFieldChange;
353
354        tester!(alix);
355        tester!(bo);
356        let (dm, _) = alix.test_talk_in_dm_with(&bo).await?;
357        let query = MsgQueryArgs {
358            content_types: Some(vec![ContentType::GroupUpdated]),
359            ..Default::default()
360        };
361        let old_count = dm.find_messages_v2(&query)?.len();
362        let metadata = GroupUpdated {
363            metadata_field_changes: vec![MetadataFieldChange {
364                field_name: "expiration_setting".into(),
365                old_value: None,
366                new_value: Some("123".into()),
367            }],
368            ..Default::default()
369        };
370        let mut rows = vec![encode(&metadata); 3];
371        rows.extend(vec![
372            encode(&add_update("repeated"));
373            PAGE_SIZE as usize + 25
374        ]);
375        seed_tied_updates(&alix.db(), dm.group_id, rows);
376        let before = dm.find_messages_v2(&query)?;
377        assert_eq!(before.len(), old_count + PAGE_SIZE as usize + 28);
378        alix.db().raw_query(|conn| {
379            xmtp_db::diesel::update(xmtp_db::schema::user_preferences::table)
380                .set(xmtp_db::schema::user_preferences::dm_group_updates_migrated.eq(false))
381                .execute(conn)
382        })?;
383        let _reopened = crate::builder::ClientBuilder::from_client(alix.client.clone())
384            .with_disable_workers(true)
385            .build()
386            .await?;
387        // Bound the observation: startup must no longer complete this migration.
388        let _ = xmtp_common::time::timeout(
389            std::time::Duration::from_secs(1),
390            xmtp_common::wait_for_eq(
391                || async {
392                    StoredUserPreferences::load(alix.db())
393                        .unwrap()
394                        .dm_group_updates_migrated
395                },
396                true,
397            ),
398        )
399        .await;
400        let after = dm.find_messages_v2(&query)?;
401        assert_eq!(
402            after.len(),
403            before.len(),
404            "startup must preserve all DM update history"
405        );
406        for message in before {
407            assert!(
408                after
409                    .iter()
410                    .any(|row| row.metadata.id == message.metadata.id)
411            );
412        }
413    }
414
415    /// Rows written in one millisecond share `inserted_at_ns`, and the cursor
416    /// is a strict `>` on that column. A tie group larger than one page must
417    /// still be read in full, or the newest update is never consumed and the
418    /// duplicate check answers from stale rows.
419    #[xmtp_common::test(unwrap_try = true)]
420    async fn test_update_already_exists_spans_rows_sharing_one_inserted_at() {
421        tester!(alix);
422        tester!(bo);
423        let (dm, _) = alix.test_talk_in_dm_with(&bo).await?;
424
425        let earlier = add_update("earlier");
426        let latest = add_update("latest");
427        let mut rows = vec![encode(&earlier); PAGE_SIZE as usize];
428        rows.extend(vec![encode(&latest); 25]);
429        seed_tied_updates(&alix.db(), dm.group_id, rows);
430
431        let storage = alix.context.mls_storage();
432        assert!(
433            dm.update_already_exists(&latest, storage)?,
434            "the newest add sits past the page edge and must still be seen"
435        );
436        assert!(
437            !dm.update_already_exists(&earlier, storage)?,
438            "an add that is no longer the newest is not a duplicate"
439        );
440    }
441
442    /// The page count must come from the rows the database returned, not from
443    /// the rows that decoded. A row that fails to decode would otherwise make a
444    /// full page look short, so its trailing tie group is consumed as if the
445    /// scan had ended, and the rows past the page edge are skipped.
446    #[xmtp_common::test(unwrap_try = true)]
447    async fn test_update_already_exists_counts_undecodable_rows_toward_the_page() {
448        tester!(alix);
449        tester!(bo);
450        let (dm, _) = alix.test_talk_in_dm_with(&bo).await?;
451
452        let earlier = add_update("earlier");
453        let latest = add_update("latest");
454        // Invalid protobuf, inside the first page.
455        let mut rows = vec![vec![0xFF, 0xFE, 0xFD]];
456        rows.extend(vec![encode(&earlier); PAGE_SIZE as usize - 1]);
457        rows.extend(vec![encode(&latest); 25]);
458        seed_tied_updates(&alix.db(), dm.group_id, rows);
459
460        let storage = alix.context.mls_storage();
461        assert!(
462            dm.update_already_exists(&latest, storage)?,
463            "an undecodable row must not shorten the page and hide the rows past its edge"
464        );
465    }
466}
xmtp_mls/groups/mls_sync/receive.rs

xmtp_mls/groups/mls_sync/
receive.rs
