xmtp_mls_core/
sync_tasks.rs

1//! Durable sync tasks that Welcome processing and device sync enqueue
2//! inside their own storage transactions.
3
4use crate::events::InternalEvent;
5use xmtp_common::{NS_IN_DAY, time::now_ns};
6use xmtp_configuration::SYNC_UPDATE_INSTALLATIONS_INTERVAL_NS;
7use xmtp_db::tasks::NewTask;
8use xmtp_db::{StorageError, group::ConversationType, prelude::*};
9use xmtp_events::EventWriter;
10use xmtp_proto::types::GroupId;
11use xmtp_proto::xmtp::mls::database::{
12    AddMissingInstallations as AddMissingInstallationsProto, Task as TaskProto,
13    task::Task as TaskKindProto,
14};
15
16/// Maximum duplicate groups per DM scheduled in addition to its winner.
17const MAX_DM_EXTRA_GROUPS: usize = 10;
18
19/// A new sync group from a Welcome means a new installation: schedule one
20/// durable task per eligible group, with bounded DM extras (deduped by
21/// payload hash), in the transaction that installs the Welcome, so a restart
22/// cannot lose them.
23// implements: SYNC-014
24pub fn schedule_newcomer_adds(
25    db: &impl DbQuery,
26    events: &impl EventWriter<InternalEvent>,
27) -> Result<usize, StorageError> {
28    use xmtp_db::consent_record::ConsentState;
29    use xmtp_db::diesel::{RunQueryDsl, sql_query, sql_types::*};
30    use xmtp_db::group::{GroupMembershipState::*, StoredGroup, consent_join_sql};
31
32    let consent = [ConsentState::Allowed, ConsentState::Unknown].map(|s| s as i32);
33    let membership = [Allowed, Rejected, Pending, PendingRemove].map(|s| s as i32);
34    let types = ConversationType::virtual_types();
35    let virtuals: Vec<_> = types.into_iter().map(|t| t as i32).collect();
36    // implements: ARCH-015
37    let groups = db.raw_query(|conn| {
38        sql_query(format!(
39            "WITH eligible AS (
40              SELECT g.id, g.dm_id, g.last_message_ns, g.created_at_ns FROM groups g LEFT JOIN consent_records ON {}
41              WHERE (state IS NULL OR state IN (SELECT value FROM json_each(?))) AND COALESCE(last_message_ns, created_at_ns) > ?
42                AND membership_state IN (SELECT value FROM json_each(?)) AND conversation_type NOT IN (SELECT value FROM json_each(?))
43            ), winners AS (
44              SELECT *, ROW_NUMBER() OVER (PARTITION BY dm_id ORDER BY
45                COALESCE(last_message_ns, 0) DESC, id DESC) AS winner_rank FROM eligible WHERE dm_id IS NOT NULL
46            ), ranked AS (
47              SELECT id, ROW_NUMBER() OVER (PARTITION BY dm_id ORDER BY
48                winner_rank = 1 DESC, COALESCE(last_message_ns, created_at_ns) DESC, id DESC) AS rank FROM winners
49            ) SELECT * FROM groups WHERE id IN (SELECT id FROM eligible WHERE dm_id IS NULL
50              UNION ALL SELECT id FROM ranked WHERE rank <= ?)",
51            consent_join_sql("g"),
52        ))
53        .bind::<Text, _>(serde_json::json!(consent).to_string())
54        .bind::<BigInt, _>(now_ns() - NS_IN_DAY * 90)
55        .bind::<Text, _>(serde_json::json!(membership).to_string())
56        .bind::<Text, _>(serde_json::json!(virtuals).to_string())
57        .bind::<BigInt, _>((MAX_DM_EXTRA_GROUPS + 1) as i64)
58        .load::<StoredGroup>(conn)
59    })?;
60    for group in &groups {
61        schedule_add_missing_installations(db, events, group.id, None, false)?;
62    }
63    Ok(groups.len())
64}
65
66/// Durably enqueue one AddMissingInstallations task for `group_id` (deduped
67/// by payload hash against any pending row for the same group) and wake the
68/// TaskRunner. The task is due once `delay`, a rate-limit delay, has passed.
69/// A task `until_done` never expires, and a failing one backs off to at most
70/// one run an interval.
71pub fn schedule_add_missing_installations(
72    db: &impl DbQuery,
73    events: &impl EventWriter<InternalEvent>,
74    group_id: GroupId,
75    delay: Option<xmtp_common::time::Duration>,
76    until_done: bool,
77) -> Result<(), StorageError> {
78    let due = now_ns().saturating_add(delay.map_or(0, |delay| {
79        i64::try_from(delay.as_nanos()).unwrap_or(i64::MAX)
80    }));
81    let mut task = NewTask::builder();
82    task.originating_message_sequence_id(0)
83        .next_attempt_at_ns(due);
84    if until_done {
85        task.expires_at_ns(i64::MAX)
86            .max_attempts(i32::MAX)
87            .max_backoff_duration_ns(SYNC_UPDATE_INSTALLATIONS_INTERVAL_NS);
88    } else {
89        // The usual three days to succeed count from when the task is due.
90        task.expires_at_ns(due.saturating_add(NS_IN_DAY * 3));
91    }
92    let task = task.build(TaskProto {
93        task: Some(TaskKindProto::AddMissingInstallations(
94            AddMissingInstallationsProto {
95                group_id: group_id.to_vec(),
96            },
97        )),
98    })?;
99    let hash = task.data_hash.clone();
100    db.create_or_ignore_task(task)?;
101    if until_done {
102        // A row an older build stored for this group is kept instead of the
103        // new one, so it must not expire either.
104        use xmtp_db::{diesel, diesel::prelude::*, schema::tasks::dsl};
105        db.raw_query(|c| {
106            diesel::update(dsl::tasks.filter(dsl::data_hash.eq(&hash)))
107                .set((
108                    dsl::expires_at_ns.eq(i64::MAX),
109                    dsl::max_attempts.eq(i32::MAX),
110                    dsl::max_backoff_duration_ns.eq(SYNC_UPDATE_INSTALLATIONS_INTERVAL_NS),
111                    dsl::next_attempt_at_ns.eq(diesel::dsl::sql::<diesel::sql_types::BigInt>(
112                        "MAX(next_attempt_at_ns, ",
113                    )
114                    .bind::<diesel::sql_types::BigInt, _>(due)
115                    .sql(")")),
116                ))
117                .execute(c)
118        })?;
119    }
120    events.emit(None, Some(InternalEvent::TaskScheduled));
121    Ok(())
122}
xmtp_mls_core/sync_tasks.rs

xmtp_mls_core/
sync_tasks.rs
