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            expire_at_ns: None,
162            inserted_at_ns: 0, // Will be set by database
163            should_push: true,
164            // Matches the key used to derive `message_id` above.
165            idempotency_key: timestamp_ns.to_string(),
166        };
167
168        msg.store_or_ignore(&storage.db())?;
169        Ok(Some((msg, payload)))
170    }
171
172    fn update_already_exists(
173        &self,
174        payload: &GroupUpdated,
175        storage: &impl XmtpMlsStorageProvider,
176    ) -> Result<bool, GroupMessageProcessingError> {
177        if self.dm_id.is_none() || payload.added_inboxes.is_empty() {
178            // Only dedupe for DMs.
179            // Only dedupe for group adds.
180            return Ok(false);
181        }
182
183        let mut deduper = GroupUpdateDeduper::default();
184        let mut inserted_after_ns = None;
185        // Grows while a page holds a single `inserted_at_ns`, so that a tie
186        // group larger than one page still fits and the cursor can advance.
187        let mut limit = PAGE_SIZE;
188        loop {
189            // DMs are stitched, so we don't want to have the same
190            // group updates from multiple DMs being saved to the database.
191            //
192            // Sort by the same column the cursor advances on. The default sort
193            // is `SentAt`, which does not agree with an `inserted_at_ns`
194            // cursor: the last row by sent time need not hold the largest
195            // `inserted_at_ns`, so the cursor could stall and repeat a page
196            // forever.
197            //
198            // Read stored rows, not decoded messages. The page count must be
199            // the number of rows the database returned: a decoded view drops
200            // rows it cannot decode, which makes a full page look short and
201            // lets the rows past its edge be skipped. Decoding one row at a
202            // time also skips the reaction and reply lookups that a decoded
203            // list runs for every page, which group updates never need.
204            let rows = storage.db().get_group_messages(
205                &self.group_id,
206                &MsgQueryArgs {
207                    content_types: Some(vec![ContentType::GroupUpdated]),
208                    inserted_after_ns,
209                    limit: Some(limit),
210                    sort_by: Some(SortBy::InsertedAt),
211                    ..Default::default()
212                },
213            )?;
214
215            let full_page = rows.len() >= limit as usize;
216            let last_value = rows.last().map(|row| row.inserted_at_ns);
217
218            // The cursor filter is a strict `>` on `inserted_at_ns`, which the
219            // database records at millisecond granularity, so rows written in
220            // the same millisecond share a value. On a full page the final
221            // value may continue past the page edge, so hold those rows back
222            // and let the next page re-read them in full. A short page reached
223            // the end of the scan and keeps every row.
224            let cutoff = last_value.filter(|_| full_page);
225            let consumed = rows
226                .into_iter()
227                .filter(|row| Some(row.inserted_at_ns) != cutoff);
228
229            let mut highest_consumed = None;
230            for row in consumed {
231                highest_consumed = Some(row.inserted_at_ns);
232                let MessageBody::GroupUpdated(update) = DecodedMessage::from(row).content else {
233                    continue;
234                };
235
236                deduper.consume(&update);
237            }
238
239            match highest_consumed {
240                // Resume just above the last value read in full.
241                Some(value) => {
242                    inserted_after_ns = Some(value);
243                    limit = PAGE_SIZE;
244                }
245                // A full page of one repeated value: nothing could be retired,
246                // so re-read it larger rather than stall.
247                None if full_page => limit = limit.saturating_mul(2),
248                None => break,
249            }
250        }
251
252        Ok(deduper.is_dupe(payload))
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259    use crate::tester;
260    use xmtp_common::{rand_vec, time::now_ns};
261    use xmtp_content_types::{ContentCodec, encryption::sha256, group_updated::GroupUpdatedCodec};
262    use xmtp_db::diesel::{ExpressionMethods, QueryDsl, RunQueryDsl};
263    use xmtp_db::group_message::{DeliveryStatus, GroupMessageKind};
264    use xmtp_db::user_preferences::StoredUserPreferences;
265    use xmtp_db::{ConnectionExt, DbConnection, Store};
266    use xmtp_proto::types::GroupId;
267    use xmtp_proto::xmtp::mls::message_contents::group_updated::Inbox;
268
269    /// Marks the rows a test seeds, so they can be pinned to one `inserted_at_ns`.
270    const SEEDED_SENDER: &str = "seeded";
271
272    fn add_update(inbox_id: &str) -> GroupUpdated {
273        GroupUpdated {
274            added_inboxes: vec![Inbox {
275                inbox_id: inbox_id.to_string(),
276            }],
277            ..Default::default()
278        }
279    }
280
281    fn encode(update: &GroupUpdated) -> Vec<u8> {
282        let mut bytes = Vec::new();
283        GroupUpdatedCodec::encode(update.clone())
284            .unwrap()
285            .encode(&mut bytes)
286            .unwrap();
287        bytes
288    }
289
290    /// Store `rows` as group updates of `group_id`, in order, then pin them to
291    /// one `inserted_at_ns` one millisecond above every row the group already
292    /// holds.
293    fn seed_tied_updates<C: ConnectionExt>(
294        db: &DbConnection<C>,
295        group_id: GroupId,
296        rows: Vec<Vec<u8>>,
297    ) {
298        for (i, bytes) in rows.into_iter().enumerate() {
299            StoredGroupMessage {
300                id: sha256(&rand_vec::<12>()),
301                group_id,
302                decrypted_message_bytes: bytes,
303                sent_at_ns: now_ns(),
304                kind: GroupMessageKind::MembershipChange,
305                sender_installation_id: vec![1, 2, 3],
306                sender_inbox_id: SEEDED_SENDER.to_string(),
307                delivery_status: DeliveryStatus::Published,
308                content_type: ContentType::GroupUpdated,
309                version_major: 0,
310                version_minor: 0,
311                authority_id: "unknown".to_string(),
312                reference_id: None,
313                sequence_id: i as i64 + 1,
314                envelope_hash: None,
315                expiry_ns: None,
316                expire_at_ns: None,
317                inserted_at_ns: 0,
318                should_push: true,
319                idempotency_key: String::new(),
320            }
321            .store(db)
322            .unwrap();
323        }
324
325        // The database stamps `inserted_at_ns` at insert time, and its clock
326        // does not agree with `now_ns`. Pin the seeded rows to one value just
327        // above every row the group already holds, rather than rely on how
328        // fast the loop ran or on the wall clock.
329        db.raw_query(|conn| {
330            use xmtp_db::schema::group_messages::dsl;
331            let newest: Option<i64> = dsl::group_messages
332                .filter(dsl::group_id.eq(group_id))
333                .select(xmtp_db::diesel::dsl::max(dsl::inserted_at_ns))
334                .first(conn)?;
335            let shared_inserted_at_ns = newest.unwrap_or_default() + 1_000_000;
336            xmtp_db::diesel::update(dsl::group_messages)
337                .filter(dsl::group_id.eq(group_id))
338                .filter(dsl::sender_inbox_id.eq(SEEDED_SENDER))
339                .set(dsl::inserted_at_ns.eq(shared_inserted_at_ns))
340                .execute(conn)
341        })
342        .unwrap();
343    }
344
345    /// The startup migration deletes repeated group updates in DMs. Let it
346    /// finish before seeding repeated rows, so it cannot thin them out.
347    async fn wait_for_startup_cleanup<C: ConnectionExt>(db: DbConnection<C>) {
348        xmtp_common::wait_for_eq(
349            || async {
350                StoredUserPreferences::load(&db)
351                    .unwrap()
352                    .dm_group_updates_migrated
353            },
354            true,
355        )
356        .await
357        .unwrap();
358    }
359
360    /// Rows written in one millisecond share `inserted_at_ns`, and the cursor
361    /// is a strict `>` on that column. A tie group larger than one page must
362    /// still be read in full, or the newest update is never consumed and the
363    /// duplicate check answers from stale rows.
364    #[xmtp_common::test(unwrap_try = true)]
365    async fn test_update_already_exists_spans_rows_sharing_one_inserted_at() {
366        tester!(alix);
367        tester!(bo);
368        let (dm, _) = alix.test_talk_in_dm_with(&bo).await?;
369        wait_for_startup_cleanup(alix.db()).await;
370
371        let earlier = add_update("earlier");
372        let latest = add_update("latest");
373        let mut rows = vec![encode(&earlier); PAGE_SIZE as usize];
374        rows.extend(vec![encode(&latest); 25]);
375        seed_tied_updates(&alix.db(), dm.group_id, rows);
376
377        let storage = alix.context.mls_storage();
378        assert!(
379            dm.update_already_exists(&latest, storage)?,
380            "the newest add sits past the page edge and must still be seen"
381        );
382        assert!(
383            !dm.update_already_exists(&earlier, storage)?,
384            "an add that is no longer the newest is not a duplicate"
385        );
386    }
387
388    /// The page count must come from the rows the database returned, not from
389    /// the rows that decoded. A row that fails to decode would otherwise make a
390    /// full page look short, so its trailing tie group is consumed as if the
391    /// scan had ended, and the rows past the page edge are skipped.
392    #[xmtp_common::test(unwrap_try = true)]
393    async fn test_update_already_exists_counts_undecodable_rows_toward_the_page() {
394        tester!(alix);
395        tester!(bo);
396        let (dm, _) = alix.test_talk_in_dm_with(&bo).await?;
397        wait_for_startup_cleanup(alix.db()).await;
398
399        let earlier = add_update("earlier");
400        let latest = add_update("latest");
401        // Invalid protobuf, inside the first page.
402        let mut rows = vec![vec![0xFF, 0xFE, 0xFD]];
403        rows.extend(vec![encode(&earlier); PAGE_SIZE as usize - 1]);
404        rows.extend(vec![encode(&latest); 25]);
405        seed_tied_updates(&alix.db(), dm.group_id, rows);
406
407        let storage = alix.context.mls_storage();
408        assert!(
409            dm.update_already_exists(&latest, storage)?,
410            "an undecodable row must not shorten the page and hide the rows past its edge"
411        );
412    }
413}
xmtp_mls/groups/mls_sync/receive.rs

xmtp_mls/groups/mls_sync/
receive.rs
