diff --git a/crates/xmtp_db/src/encrypted_store/tasks.rs b/crates/xmtp_db/src/encrypted_store/tasks.rs index a2038319b7..d317680f4f 100644 --- a/crates/xmtp_db/src/encrypted_store/tasks.rs +++ b/crates/xmtp_db/src/encrypted_store/tasks.rs @@ -550,7 +550,9 @@ pub(crate) mod tests { /// constants without a row-migration story. #[xmtp_common::test] fn data_hash_encoding_is_pinned() { - use xmtp_proto::xmtp::mls::database::{KpDeletion, KpRotation, PullInDeadline}; + use xmtp_proto::xmtp::mls::database::{ + AddMissingInstallations, KpDeletion, KpRotation, PullInDeadline, + }; let rotation = TaskProto { task: Some(TaskKind::KpRotation(KpRotation {})), }; @@ -579,6 +581,19 @@ pub(crate) mod tests { hex::encode(data_hash_for(&pull_in)), "16b424873a34096e5157ab9f0a31e80dff1d23ebd8b1aab2b948a4732abfc849" ); + // AddMissingInstallations (bytes group_id): outer tag for oneof field 7 + // is (7<<3)|2 = 0x3a, LEN 0x22 (34 = 2-byte inner header + 32 bytes); + // inner field 1 tag 0x0a, LEN 0x20. + let add = TaskProto { + task: Some(TaskKind::AddMissingInstallations(AddMissingInstallations { + group_id: vec![0x22; 32], + })), + }; + assert_eq!(add.encode_to_vec()[..4], [0x3a, 0x22, 0x0a, 0x20]); + assert_eq!( + hex::encode(data_hash_for(&add)), + "2180ffa08703d0dcc600a070da9d162da6821e29176af24e7a9955af2cf43764" + ); // Determinism under repetition and a decode round-trip. let bytes = pull_in.encode_to_vec(); for _ in 0..100 { diff --git a/crates/xmtp_mls/src/groups/intents/queue.rs b/crates/xmtp_mls/src/groups/intents/queue.rs index b4c48b83d4..3419c7db8f 100644 --- a/crates/xmtp_mls/src/groups/intents/queue.rs +++ b/crates/xmtp_mls/src/groups/intents/queue.rs @@ -1,11 +1,6 @@ -use futures::{StreamExt, TryFutureExt, stream}; -use std::collections::HashSet; -use std::future::Future; - use crate::groups::intents::GROUP_KEY_ROTATION_INTERVAL_NS; use crate::groups::{GroupError, MlsGroup, XmtpSharedContext}; use derive_builder::Builder; -use itertools::Itertools; use xmtp_db::TransactionOutcome::Continue; use xmtp_db::{ DbQuery, TransactionOutcome, @@ -41,71 +36,6 @@ impl QueueIntentBuilder { }) .map(TransactionOutcome::into_continued) } - - /// Queue this intent for each group - /// Accepts a hashset to ensure each group is unique - /// Accepts a closure that returns the intent data for a group - pub async fn queue_for_each<'a, C, F, Fut, E>( - &mut self, - groups: HashSet>, - data: F, - ) -> Result, GroupError> - where - C: XmtpSharedContext + 'a, - F: Fn(MlsGroup) -> Fut, - Fut: Future, E>>, - GroupError: From, - E: std::fmt::Debug + std::error::Error, - { - if groups.is_empty() { - return Ok(vec![]); - } - - let groups: Vec> = Vec::from_iter(groups); - let context: C = groups.first().expect("checked for empty").context.clone(); - - // get the intent data for each group - let (groups, errors): (Vec<_>, Vec<_>) = stream::iter(groups) - .map(|group| data(group.clone()).map_ok(move |d| (group, d))) - .buffered(10) - .collect::>>() - .await - .into_iter() - .partition_result(); - let mut errors = errors.into_iter().map(GroupError::from).collect::>(); - - let (intents, errs): (Vec, Vec<_>) = context - .mls_storage() - .transaction(|conn| { - let (intents, errs): (Vec, Vec<_>) = groups - .into_iter() - .map(|(group, data)| { - let storage = conn.key_store(); - - // nesting a transaction uses SQLite Savepoints - // https://sqlite.org/lang_savepoint.html - // Each savepoint here always commits or errors, so - // unwrap the committed value directly. - storage - .savepoint(|conn| { - let intent = self.clone().data(data).build()?; - let storage = conn.key_store(); - let db = storage.db(); - intent.queue_with_conn(&db, &group).map(Continue) - }) - .map(TransactionOutcome::into_continued) - }) - .partition_result(); - Ok::<_, GroupError>(Continue((intents, errs))) - })? - .into_continued(); - errors.extend(errs); - - for error in errors { - tracing::warn!("failed to queue intent {error}"); - } - Ok(intents) - } } impl QueueIntent { @@ -259,67 +189,3 @@ impl QueueIntent { Ok(()) } } - -#[cfg(test)] -mod tests { - use std::{iter, sync::Arc}; - use tokio::sync::Mutex; - use xmtp_common::Generate; - use xmtp_db::group::{GroupMembershipState, StoredGroup}; - use xmtp_proto::types::GroupId; - - use crate::test::mock::{NewMockContext, context}; - - use super::*; - use rstest::*; - - fn group(context: &C, id: Option) -> MlsGroup<&C> { - let id = id.unwrap_or_else(GroupId::generate); - StoredGroup::builder() - .id(id) - .created_at_ns(1) - .membership_state(GroupMembershipState::Allowed) - .added_by_inbox_id("bob") - .should_publish_commit_log(false) - .build() - .unwrap() - .store(&context.mls_storage().db()) - .unwrap(); - - MlsGroup { - group_id: id, - dm_id: None, - created_at_ns: 1, - context, - mls_commit_lock: Arc::new(Default::default()), - mutex: Arc::new(Mutex::new(())), - conversation_type: xmtp_db::group::ConversationType::Dm, - } - } - - #[rstest] - #[xmtp_common::test] - async fn can_queue_intent_for_each_group(context: NewMockContext) { - let groups = iter::repeat_with(|| group(&context, None)).take(10); - let intents = QueueIntent::update_group_membership() - .data(vec![0, 1, 2]) - .queue_for_each(groups.collect(), async |_| Ok::<_, GroupError>(vec![])) - .await - .unwrap(); - assert_eq!(intents.len(), 10); - } - - #[rstest] - #[xmtp_common::test] - async fn only_queues_for_unique(context: NewMockContext) { - let groups = iter::repeat_n(group(&context, Some(GroupId::ZERO)), 10); - - let stored = QueueIntent::update_group_membership() - .data(vec![0, 1, 2]) - .queue_for_each(groups.collect(), async |_g| Ok::<_, GroupError>(vec![])) - .await - .unwrap(); - - assert_eq!(stored.len(), 1); - } -} diff --git a/crates/xmtp_mls/src/test/client_test_utils.rs b/crates/xmtp_mls/src/test/client_test_utils.rs index 81d507070a..13e8b8043a 100644 --- a/crates/xmtp_mls/src/test/client_test_utils.rs +++ b/crates/xmtp_mls/src/test/client_test_utils.rs @@ -73,14 +73,20 @@ where let other_sync = other.device_sync_client(); let mut sync_group = this_sync.get_sync_group().await?; let mut other_sync_group = other_sync.get_sync_group().await?; - for i in 0..10 { + for _ in 0..10 { sync_group = this_sync.get_sync_group().await?; other_sync_group = other_sync.get_sync_group().await?; sync_group.sync().await?; other_sync_group.sync().await?; - if sync_group.group_id == other_sync_group.group_id { + // Converge on epoch too, not just the group id: a commit published + // right after one side joins (e.g. the post-join KeyUpdate that + // rides the first HMAC-cycle message) is in flight exactly when + // the ids first match. + if sync_group.group_id == other_sync_group.group_id + && sync_group.epoch().await? == other_sync_group.epoch().await? + { break; } } diff --git a/crates/xmtp_mls/src/worker/device_sync/mod.rs b/crates/xmtp_mls/src/worker/device_sync/mod.rs index 926e1ec1e6..116a2f52f3 100644 --- a/crates/xmtp_mls/src/worker/device_sync/mod.rs +++ b/crates/xmtp_mls/src/worker/device_sync/mod.rs @@ -2,20 +2,16 @@ use crate::{ client::ClientError, context::XmtpSharedContext, groups::{ - GroupError, MlsGroup, PreconfiguredPolicies, intents::QueueIntent, send_message_opts, - summary::SyncSummary, welcome_sync::WelcomeService, + GroupError, MlsGroup, PreconfiguredPolicies, send_message_opts, summary::SyncSummary, + welcome_sync::WelcomeService, }, mls_store::{MlsStore, MlsStoreError}, subscriptions::{SubscribeError, SyncWorkerEvent}, worker::{NeedsDbReconnect, metrics::WorkerMetrics}, }; -use futures::{StreamExt, TryStreamExt, stream}; use owo_colors::OwoColorize; use prost::Message; -use std::{ - collections::{HashMap, HashSet}, - sync::Arc, -}; +use std::{collections::HashMap, sync::Arc}; use thiserror::Error; use tokio::sync::broadcast::error::RecvError; use tracing::instrument; @@ -24,6 +20,7 @@ use xmtp_archive::{ArchiveError, BackupMetadata}; use xmtp_common::ErrorCode; use xmtp_common::{NS_IN_DAY, RetryableError, time::now_ns}; use xmtp_content_types::encoded_content_to_bytes; +use xmtp_db::tasks::NewTask; use xmtp_db::{ NotFound, StorageError, consent_record::ConsentState, group::GroupQueryArgs, group_message::StoredGroupMessage, @@ -31,14 +28,20 @@ use xmtp_db::{ use xmtp_db::{XmtpDb, group::ConversationType, prelude::*}; use xmtp_id::{InboxIdRef, associations::DeserializationError}; use xmtp_mls_common::group::GroupMetadataOptions; -use xmtp_proto::types::InstallationId; +use xmtp_proto::types::{GroupId, InstallationId}; use xmtp_proto::xmtp::{ device_sync::content::{ DeviceSyncContent as DeviceSyncContentProto, device_sync_content::Content as ContentProto, }, - mls::message_contents::{ - ContentTypeId, EncodedContent, PlaintextEnvelope, - plaintext_envelope::{Content, V1}, + mls::{ + database::{ + AddMissingInstallations as AddMissingInstallationsProto, Task as TaskProto, + task::Task as TaskKindProto, + }, + message_contents::{ + ContentTypeId, EncodedContent, PlaintextEnvelope, + plaintext_envelope::{Content, V1}, + }, }, }; @@ -364,7 +367,17 @@ where hex::encode(self.context.installation_id()), hex::encode(sync_group.group_id) ); - sync_group.add_missing_installations().await?; + if let Err(inline_err) = sync_group.add_missing_installations().await { + // The group row is already persisted, so this add is never + // re-attempted (later calls take the `Some` branch) — arm + // the durable reconcile task so the TaskRunner heals it, + // then surface the original error. Armed only on failure: + // enqueue-first duplicated the reconcile (and its identity + // fetch) on every sync-group creation. + self.schedule_add_missing_installations_task(sync_group.group_id) + .map_err(Box::new)?; + return Err(inline_err); + } sync_group.sync_with_conn().await?; self.metrics.increment_metric(SyncMetric::SyncGroupCreated); @@ -378,36 +391,47 @@ where /// This should be triggered when a new sync group appears, /// indicating the presence of a new installation. + /// + /// Schedules one durable AddMissingInstallations task per eligible group + /// on the TaskRunner (deduped by payload hash) so the membership add is + /// retried with backoff instead of being lost if a single inline attempt + /// fails (e.g. identity propagation lag → MissingSequenceId). #[cfg_attr( any(test, feature = "test-utils"), tracing::instrument(level = "info", skip_all) )] - pub async fn add_new_installation_to_groups(&self) -> Result<(), DeviceSyncError> { + pub fn schedule_add_installations_to_groups(&self) -> Result { let groups = self.mls_store.find_groups(GroupQueryArgs { last_activity_after_ns: Some(now_ns() - NS_IN_DAY * 90), consent_states: Some(vec![ConsentState::Allowed, ConsentState::Unknown]), ..Default::default() })?; - let groups = HashSet::from_iter(groups); - let intents = QueueIntent::update_group_membership() - .queue_for_each(groups, move |group| async move { - let intent = group.get_membership_update_intent(&[], &[]).await?; - let intent: Vec = intent.into(); - Ok::<_, GroupError>(intent) - }) - .await?; - - let context = &self.context; - stream::iter(intents) - .map(Ok::<_, GroupError>) - .try_for_each_concurrent(10, |intent| async move { - let (group, _) = MlsGroup::new_cached(context, &intent.group_id)?; - group.sync_until_intent_resolved(intent.id).await?; - Ok(()) - }) - .await?; + for group in &groups { + self.schedule_add_missing_installations_task(group.group_id)?; + } + Ok(groups.len()) + } + /// Durably enqueue one AddMissingInstallations task for `group_id` + /// (deduped by payload hash against any pending row for the same group) + /// and wake the TaskRunner. + pub(crate) fn schedule_add_missing_installations_task( + &self, + group_id: GroupId, + ) -> Result<(), DeviceSyncError> { + let task = NewTask::builder() + .originating_message_sequence_id(0) + .originating_message_originator_id(0) + .build(TaskProto { + task: Some(TaskKindProto::AddMissingInstallations( + AddMissingInstallationsProto { + group_id: group_id.to_vec(), + }, + )), + })?; + self.context.db().create_or_ignore_task(task)?; + self.context.task_channels().wake(); Ok(()) } } diff --git a/crates/xmtp_mls/src/worker/device_sync/tests.rs b/crates/xmtp_mls/src/worker/device_sync/tests.rs index 790f4a7b59..c8522a4dc4 100644 --- a/crates/xmtp_mls/src/worker/device_sync/tests.rs +++ b/crates/xmtp_mls/src/worker/device_sync/tests.rs @@ -352,24 +352,27 @@ async fn test_only_added_to_correct_groups() { .register_interest(SyncMetric::SyncGroupWelcomesProcessed, 1) .wait() .await?; - alix2.sync_welcomes().await?; - - // Added to new fresh group - let alix2_new_group = alix2.group(&new_group.group_id); - assert!(alix2_new_group.is_ok()); + // The adds are durable TaskRunner work now — the metric above fires when + // the tasks are scheduled, not when the commits land. Poll until alix2 + // has been welcomed into every eligible group: the new fresh group, the + // unknown-consent group, and the consented DM. + xmtp_common::wait_for_some(|| async { + let _ = alix2.sync_welcomes().await; + (alix2.group(&new_group.group_id).is_ok() + && alix2.group(&alix_bo_group_unknown.group_id).is_ok() + && alix2.group(&alix_bo_dm.group_id).is_ok()) + .then_some(()) + }) + .await + .expect("alix2 must be added to all eligible groups"); + + // The negatives are meaningful once the positive set has converged: the + // filtered-out groups never get an AddMissingInstallations task at all. // Not added to old stale group let alix2_old_group = alix2.group(&old_group.group_id); assert!(alix2_old_group.is_err()); - // Added to group with unknown consent state - let alix2_bo_group_unknown = alix2.group(&alix_bo_group_unknown.group_id); - assert!(alix2_bo_group_unknown.is_ok()); - - // Added to consented DM - let alix2_bo_dm = alix2.group(&alix_bo_dm.group_id); - assert!(alix2_bo_dm.is_ok()); - // Not added to denied group from Bo let alix2_bo_group_denied = alix2.group(&alix_bo_group_denied.group_id); assert!(alix2_bo_group_denied.is_err()); @@ -496,3 +499,125 @@ async fn test_incremental_consent() { let dm2 = alix2.group(&dm.group_id)?; assert_eq!(dm2.consent_state()?, ConsentState::Allowed); } + +#[xmtp_common::timeout(std::time::Duration::from_secs(60))] +#[rstest::rstest] +#[xmtp_common::test(unwrap_try = true)] +#[cfg_attr(target_arch = "wasm32", ignore)] +async fn test_task_runner_adds_new_installation_to_groups() { + // Live sync worker + live TaskRunner (tester! defaults enable the runner). + // `stream` keeps alix1 receiving welcomes so the sync-group welcome from + // alix2's registration reaches alix1's device-sync worker. + tester!(alix1, stream, sync_worker); + tester!(bo); + + let group = alix1 + .create_group_with_members(&[bo.inbox_id()], None, None) + .await?; + + tester!(alix2, from: alix1); + + // The welcome handler enqueues the task; the TaskRunner publishes the + // membership commit; alix2 then receives the group via welcome. Poll — + // the whole chain is async. + xmtp_common::wait_for_some(|| async { + let _ = alix2.sync_welcomes().await; + alix2.group(&group.group_id).ok().map(|_| ()) + }) + .await + .expect("alix2 must receive the group via the TaskRunner membership add"); +} + +#[xmtp_common::timeout(std::time::Duration::from_secs(30))] +#[rstest::rstest] +#[xmtp_common::test(unwrap_try = true)] +#[cfg_attr(target_arch = "wasm32", ignore)] +async fn test_sync_group_creation_leaves_no_reconcile_task() { + use crate::worker::{WorkerConfig, WorkerKind}; + use prost::Message; + use xmtp_db::tasks::QueryTasks; + use xmtp_proto::xmtp::mls::database::{Task as TaskProto, task::Task as TaskKind}; + + // TaskRunner disabled so any enqueued row would be observable (not consumed). + let mut cfg = WorkerConfig::default(); + cfg.enabled.insert(WorkerKind::TaskRunner, false); + tester!(alix, worker_config: cfg); + + alix.device_sync_client().get_sync_group().await?; + + // The durable reconcile task is armed only from the inline add's error + // path. A successful creation must NOT leave one behind — an enqueue-first + // version duplicated the reconcile (and its identity fetch) on every + // sync-group creation, breaking the pinned network-call-count tests on + // mobile bindings. + let has_task = alix.context.db().get_tasks()?.iter().any(|t| { + matches!( + TaskProto::decode(t.data.as_slice()) + .ok() + .and_then(|p| p.task), + Some(TaskKind::AddMissingInstallations(_)) + ) + }); + assert!( + !has_task, + "successful sync-group creation must not enqueue a reconcile task" + ); +} + +#[xmtp_common::timeout(std::time::Duration::from_secs(30))] +#[rstest::rstest] +#[xmtp_common::test(unwrap_try = true)] +#[cfg_attr(target_arch = "wasm32", ignore)] +async fn test_welcome_schedules_add_installation_tasks() { + use crate::worker::{WorkerConfig, WorkerKind}; + use prost::Message; + use xmtp_db::tasks::QueryTasks; + use xmtp_proto::xmtp::mls::database::{Task as TaskProto, task::Task as TaskKind}; + + // TaskRunner disabled so enqueued rows are observable (not consumed). + let mut cfg = WorkerConfig::default(); + cfg.enabled.insert(WorkerKind::TaskRunner, false); + tester!(alix1, worker_config: cfg); + tester!(bo); + + let group = alix1 + .create_group_with_members(&[bo.inbox_id()], None, None) + .await?; + + let count_add_tasks = || -> Vec> { + alix1 + .context + .db() + .get_tasks() + .unwrap() + .iter() + .filter_map(|t| match TaskProto::decode(t.data.as_slice()).ok()?.task { + Some(TaskKind::AddMissingInstallations(a)) => Some(a.group_id), + _ => None, + }) + .collect() + }; + + // Call the schedule path directly (unit level — no live sync worker needed). + let scheduled = alix1 + .device_sync_client() + .schedule_add_installations_to_groups()?; + assert!(scheduled >= 1); + + let add_tasks = count_add_tasks(); + assert!( + add_tasks.iter().any(|gid| gid == &group.group_id.to_vec()), + "expected an AddMissingInstallations task for the conversation group" + ); + + // Re-scheduling dedups on payload hash: row count stays put. + alix1 + .device_sync_client() + .schedule_add_installations_to_groups()?; + let after = count_add_tasks(); + assert_eq!( + add_tasks.len(), + after.len(), + "create_or_ignore must dedup identical payloads" + ); +} diff --git a/crates/xmtp_mls/src/worker/device_sync/worker.rs b/crates/xmtp_mls/src/worker/device_sync/worker.rs index 5762426593..f8eb287567 100644 --- a/crates/xmtp_mls/src/worker/device_sync/worker.rs +++ b/crates/xmtp_mls/src/worker/device_sync/worker.rs @@ -135,7 +135,27 @@ where } async fn run_internal(&mut self) -> Result<(), DeviceSyncError> { - while let Ok(event) = self.receiver.recv().await { + use tokio::sync::broadcast::error::RecvError; + loop { + let event = match self.receiver.recv().await { + Ok(event) => event, + Err(RecvError::Lagged(skipped)) => { + // The skipped events may have included NewSyncGroupFromWelcome, + // whose durable task rows were never created. Re-scheduling is + // cheap and deduped, so recover level-triggered instead of + // losing the edge; a Tick-equivalent sweep covers skipped + // NewSyncGroupMsg events the same way. + tracing::warn!( + skipped, + "sync worker receiver lagged; re-scheduling installation reconciliation" + ); + self.client.schedule_add_installations_to_groups()?; + self.evt_new_sync_group_msg(true).await?; + continue; + } + Err(RecvError::Closed) => break, + }; + // Tick is the internal timer heartbeat (every 20s): no real work, so // dispatch it directly without opening a worker_turn span. if matches!(event, SyncWorkerEvent::Tick) { @@ -231,8 +251,9 @@ where tracing::info!("New sync group from welcome detected."); // A new sync group from a welcome indicates a new installation. - // We need to add that installation to the groups. - self.client.add_new_installation_to_groups().await?; + // Schedule durable per-group reconciliation on the TaskRunner — + // a one-shot inline add here is lost forever if it fails once. + self.client.schedule_add_installations_to_groups()?; self.metrics .increment_metric(SyncMetric::SyncGroupWelcomesProcessed); diff --git a/crates/xmtp_mls/src/worker/tasks.rs b/crates/xmtp_mls/src/worker/tasks.rs index f8554412e9..8e911f66e4 100644 --- a/crates/xmtp_mls/src/worker/tasks.rs +++ b/crates/xmtp_mls/src/worker/tasks.rs @@ -437,6 +437,9 @@ where .unwrap_or(now + KEY_PACKAGE_ROTATION_INTERVAL_NS); return Ok(TaskOutcome::RescheduleAt(next)); } + Some(xmtp_proto::xmtp::mls::database::task::Task::AddMissingInstallations(add)) => { + Self::run_add_missing_installations(task, add, context).await?; + } None => { tracing::error!("Task {} has no data. Deleting.", task.id); context.db().delete_task(task.id)?; @@ -493,6 +496,42 @@ where Err(e) => Err(e.into()), } } + + /// Run an `AddMissingInstallations` task: load the group and reconcile its + /// membership with the inbox's latest identity state. Idempotent — the + /// underlying group method no-ops when membership is already current, so + /// retries and duplicate rows are safe. + async fn run_add_missing_installations( + task: &DbTask, + add: xmtp_proto::xmtp::mls::database::AddMissingInstallations, + context: &Context, + ) -> Result<(), TaskWorkerError> { + // A malformed group_id can never succeed — drop the task, don't retry. + let Ok(group_id) = xmtp_proto::types::GroupId::try_from(add.group_id.as_slice()) else { + tracing::warn!( + "Task {} has a malformed group_id for AddMissingInstallations. Deleting.", + task.id + ); + context.db().delete_task(task.id)?; + return Ok(()); + }; + match crate::mls_store::MlsStore::new(context.clone()).group(&group_id) { + Ok(group) => { + group.add_missing_installations().await?; + Ok(()) + } + Err(crate::mls_store::MlsStoreError::NotFound(_)) => { + tracing::debug!( + "Task {} targets a group that no longer exists. Deleting.", + task.id + ); + context.db().delete_task(task.id)?; + Ok(()) + } + // A DB/connection error is transient — let it retry. + Err(e) => Err(e.into()), + } + } async fn process_welcome_pointer( task: &DbTask, welcome_pointer: xmtp_proto::xmtp::mls::message_contents::WelcomePointer, @@ -590,6 +629,26 @@ mod tests { assert!(db.get_tasks()?.is_empty(), "Done task must be deleted"); } + #[xmtp_common::test(unwrap_try = true)] + async fn add_missing_installations_missing_group_deletes_task() { + tester!(alix, worker_config: no_runner_cfg()); + let db = alix.context.db(); + let now = xmtp_common::time::now_ns(); + let proto = TaskProto { + task: Some(TaskKind::AddMissingInstallations( + xmtp_proto::xmtp::mls::database::AddMissingInstallations { + group_id: xmtp_common::rand_vec::<16>(), + }, + )), + }; + let row = seed(&db, proto, now - 1, i64::MAX, 0, 3); + TaskWorker::run_and_reschedule_task(row, &alix.context).await?; + assert!( + db.get_tasks()?.is_empty(), + "task for a nonexistent group must be deleted, not retried" + ); + } + #[xmtp_common::test(unwrap_try = true)] async fn recurring_task_advances_and_does_not_hot_loop() { tester!(alix, worker_config: no_runner_cfg()); diff --git a/crates/xmtp_proto/proto_version b/crates/xmtp_proto/proto_version index 36639408e2..82c2144cd0 100644 --- a/crates/xmtp_proto/proto_version +++ b/crates/xmtp_proto/proto_version @@ -1 +1 @@ -f5191f73adb09156e3c120d44ca9c33b88ff4dae +a9b4d56168bd22bca4f3d5a65e841d9989a91ebc diff --git a/crates/xmtp_proto/src/gen/proto_descriptor.bin b/crates/xmtp_proto/src/gen/proto_descriptor.bin index 7e1fe416d0..d41b5c57a8 100644 Binary files a/crates/xmtp_proto/src/gen/proto_descriptor.bin and b/crates/xmtp_proto/src/gen/proto_descriptor.bin differ diff --git a/crates/xmtp_proto/src/gen/xmtp.mls.database.rs b/crates/xmtp_proto/src/gen/xmtp.mls.database.rs index 2f4adec4b5..dd9656c2d6 100644 --- a/crates/xmtp_proto/src/gen/xmtp.mls.database.rs +++ b/crates/xmtp_proto/src/gen/xmtp.mls.database.rs @@ -783,7 +783,7 @@ impl PermissionPolicyOption { } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct Task { - #[prost(oneof = "task::Task", tags = "1, 2, 3, 4, 5, 6")] + #[prost(oneof = "task::Task", tags = "1, 2, 3, 4, 5, 6, 7")] pub task: ::core::option::Option, } /// Nested message and enum types in `Task`. @@ -802,6 +802,8 @@ pub mod task { KpRotation(super::KpRotation), #[prost(message, tag = "6")] KpDeletion(super::KpDeletion), + #[prost(message, tag = "7")] + AddMissingInstallations(super::AddMissingInstallations), } } impl ::prost::Name for Task { @@ -904,3 +906,24 @@ impl ::prost::Name for ProcessPendingSelfRemove { "/xmtp.mls.database.ProcessPendingSelfRemove".into() } } +/// Durable TaskRunner intent: reconcile a group's membership with the inbox's +/// latest identity state (add installations registered after the group was +/// last updated). Enqueued by the device-sync worker when a sync-group welcome +/// signals a new installation; runs on the TaskRunner with retry/backoff so a +/// transient failure (e.g. identity propagation lag) cannot permanently skip +/// the add. group_id is the target conversation. +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct AddMissingInstallations { + #[prost(bytes = "vec", tag = "1")] + pub group_id: ::prost::alloc::vec::Vec, +} +impl ::prost::Name for AddMissingInstallations { + const NAME: &'static str = "AddMissingInstallations"; + const PACKAGE: &'static str = "xmtp.mls.database"; + fn full_name() -> ::prost::alloc::string::String { + "xmtp.mls.database.AddMissingInstallations".into() + } + fn type_url() -> ::prost::alloc::string::String { + "/xmtp.mls.database.AddMissingInstallations".into() + } +} diff --git a/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs b/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs index 1e3bb9cd79..a261b03964 100644 --- a/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs +++ b/crates/xmtp_proto/src/gen/xmtp.mls.database.serde.rs @@ -290,6 +290,106 @@ impl<'de> serde::Deserialize<'de> for add_members_data::V1 { deserializer.deserialize_struct("xmtp.mls.database.AddMembersData.V1", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for AddMissingInstallations { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + use serde::ser::SerializeStruct; + let mut len = 0; + if !self.group_id.is_empty() { + len += 1; + } + let mut struct_ser = serializer.serialize_struct("xmtp.mls.database.AddMissingInstallations", len)?; + if !self.group_id.is_empty() { + #[allow(clippy::needless_borrow)] + #[allow(clippy::needless_borrows_for_generic_args)] + struct_ser.serialize_field("group_id", pbjson::private::base64::encode(&self.group_id).as_str())?; + } + struct_ser.end() + } +} +impl<'de> serde::Deserialize<'de> for AddMissingInstallations { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + "group_id", + "groupId", + ]; + + #[allow(clippy::enum_variant_names)] + enum GeneratedField { + GroupId, + __SkipField__, + } + impl<'de> serde::Deserialize<'de> for GeneratedField { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = GeneratedField; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + #[allow(unused_variables)] + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + match value { + "groupId" | "group_id" => Ok(GeneratedField::GroupId), + _ => Ok(GeneratedField::__SkipField__), + } + } + } + deserializer.deserialize_identifier(GeneratedVisitor) + } + } + struct GeneratedVisitor; + impl<'de> serde::de::Visitor<'de> for GeneratedVisitor { + type Value = AddMissingInstallations; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("struct xmtp.mls.database.AddMissingInstallations") + } + + fn visit_map(self, mut map_: V) -> std::result::Result + where + V: serde::de::MapAccess<'de>, + { + let mut group_id__ = None; + while let Some(k) = map_.next_key()? { + match k { + GeneratedField::GroupId => { + if group_id__.is_some() { + return Err(serde::de::Error::duplicate_field("groupId")); + } + group_id__ = + Some(map_.next_value::<::pbjson::private::BytesDeserialize<_>>()?.0) + ; + } + GeneratedField::__SkipField__ => { + let _ = map_.next_value::()?; + } + } + } + Ok(AddMissingInstallations { + group_id: group_id__.unwrap_or_default(), + }) + } + } + deserializer.deserialize_struct("xmtp.mls.database.AddMissingInstallations", FIELDS, GeneratedVisitor) + } +} impl serde::Serialize for AddressesOrInstallationIds { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result @@ -3108,6 +3208,9 @@ impl serde::Serialize for Task { task::Task::KpDeletion(v) => { struct_ser.serialize_field("kp_deletion", v)?; } + task::Task::AddMissingInstallations(v) => { + struct_ser.serialize_field("add_missing_installations", v)?; + } } } struct_ser.end() @@ -3132,6 +3235,8 @@ impl<'de> serde::Deserialize<'de> for Task { "kpRotation", "kp_deletion", "kpDeletion", + "add_missing_installations", + "addMissingInstallations", ]; #[allow(clippy::enum_variant_names)] @@ -3142,6 +3247,7 @@ impl<'de> serde::Deserialize<'de> for Task { PullInDeadline, KpRotation, KpDeletion, + AddMissingInstallations, __SkipField__, } impl<'de> serde::Deserialize<'de> for GeneratedField { @@ -3170,6 +3276,7 @@ impl<'de> serde::Deserialize<'de> for Task { "pullInDeadline" | "pull_in_deadline" => Ok(GeneratedField::PullInDeadline), "kpRotation" | "kp_rotation" => Ok(GeneratedField::KpRotation), "kpDeletion" | "kp_deletion" => Ok(GeneratedField::KpDeletion), + "addMissingInstallations" | "add_missing_installations" => Ok(GeneratedField::AddMissingInstallations), _ => Ok(GeneratedField::__SkipField__), } } @@ -3232,6 +3339,13 @@ impl<'de> serde::Deserialize<'de> for Task { return Err(serde::de::Error::duplicate_field("kpDeletion")); } task__ = map_.next_value::<::std::option::Option<_>>()?.map(task::Task::KpDeletion) +; + } + GeneratedField::AddMissingInstallations => { + if task__.is_some() { + return Err(serde::de::Error::duplicate_field("addMissingInstallations")); + } + task__ = map_.next_value::<::std::option::Option<_>>()?.map(task::Task::AddMissingInstallations) ; } GeneratedField::__SkipField__ => { diff --git a/sdks/js/browser-sdk/test/DeviceSync.test.ts b/sdks/js/browser-sdk/test/DeviceSync.test.ts index 8596abd71b..8e3af3f45c 100644 --- a/sdks/js/browser-sdk/test/DeviceSync.test.ts +++ b/sdks/js/browser-sdk/test/DeviceSync.test.ts @@ -1,4 +1,8 @@ -import { ConsentEntityType, ConsentState } from "@xmtp/wasm-bindings"; +import { + BackupElementSelectionOption, + ConsentEntityType, + ConsentState, +} from "@xmtp/wasm-bindings"; import { describe, expect, it, vi } from "vitest"; import { HistorySyncUrls } from "@/constants"; import { uuid } from "@/utils/uuid"; @@ -153,7 +157,13 @@ describe("DeviceSync", () => { await alix.sendSyncArchive( "123", { - elements: [], + // Request real content: an empty selection produces an empty + // archive, leaving the assertion entirely dependent on the + // membership-add welcome path. + elements: [ + BackupElementSelectionOption.Messages, + BackupElementSelectionOption.Consent, + ], excludeDisappearingMessages: false, }, HistorySyncUrls.local, @@ -214,7 +224,14 @@ describe("DeviceSync", () => { const messagesOnClient2 = await vi.waitFor(async () => { await client2.sendSyncRequest( { - elements: [], + // Request real content: with an empty selection the archive is + // empty and the group can only arrive via the membership-add + // welcome path, which converges too slowly on loaded CI runners. + // A Messages archive makes every retry attempt self-sufficient. + elements: [ + BackupElementSelectionOption.Messages, + BackupElementSelectionOption.Consent, + ], excludeDisappearingMessages: false, }, HistorySyncUrls.local, diff --git a/sdks/js/node-sdk/test/DeviceSync.test.ts b/sdks/js/node-sdk/test/DeviceSync.test.ts index 71d001d1bd..545875a087 100644 --- a/sdks/js/node-sdk/test/DeviceSync.test.ts +++ b/sdks/js/node-sdk/test/DeviceSync.test.ts @@ -1,4 +1,8 @@ -import { ConsentEntityType, ConsentState } from "@xmtp/node-bindings"; +import { + BackupElementSelectionOption, + ConsentEntityType, + ConsentState, +} from "@xmtp/node-bindings"; import { describe, expect, it, vi } from "vitest"; import { HistorySyncUrls } from "@/constants"; import { uuid } from "@/utils/uuid"; @@ -146,7 +150,13 @@ describe("DeviceSync", () => { await alix.sendSyncArchive( "123", { - elements: [], + // Request real content: an empty selection produces an empty + // archive, leaving the assertion entirely dependent on the + // membership-add welcome path. + elements: [ + BackupElementSelectionOption.Messages, + BackupElementSelectionOption.Consent, + ], excludeDisappearingMessages: false, }, HistorySyncUrls.local, @@ -210,7 +220,14 @@ describe("DeviceSync", () => { const messagesOnClient2 = await vi.waitFor(async () => { await client2.sendSyncRequest( { - elements: [], + // Request real content: with an empty selection the archive is + // empty and the group can only arrive via the membership-add + // welcome path, which converges too slowly on loaded CI runners. + // A Messages archive makes every retry attempt self-sufficient. + elements: [ + BackupElementSelectionOption.Messages, + BackupElementSelectionOption.Consent, + ], excludeDisappearingMessages: false, }, HistorySyncUrls.local,