xmtp_mls/worker/device_sync/
worker.rs

1use super::{
2    DeviceSyncClient, DeviceSyncError, ReceivedContent, decode_sync_message,
3    preference_sync::apply_preference_updates, selected_sync_group_in, with_plaintext,
4};
5pub use crate::core::metrics::SyncMetric;
6use crate::{
7    context::XmtpSharedContext,
8    identity::IdentityExt,
9    identity_updates::{
10        IdentityUpdateError, load_identity_updates_for_client, store_association_state,
11    },
12    subscriptions::{
13        incoming::{IncomingCoordinator, IncomingScope},
14        internal::{InternalEvent, emit_consent_changes, emit_root_key_changed},
15    },
16    worker::{
17        BoxedWorker, NeedsDbReconnect, NewWorker, Worker, WorkerFactory, WorkerKind, WorkerResult,
18        metrics::WorkerMetrics,
19    },
20};
21use futures::TryFutureExt;
22use std::{collections::HashSet, sync::Arc, time::Duration};
23use tokio::sync::OnceCell;
24use tracing::instrument;
25use xmtp_common::time::Instant;
26use xmtp_common::{Event, RetryableError};
27use xmtp_db::group_message::{DeliveryStatus, StoredGroupMessage};
28use xmtp_db::prelude::*;
29use xmtp_db::processed_device_sync_messages::SyncMessageKey;
30use xmtp_db::{StorageError, TransactionOutcome::Continue, sync_outbox::has_owed};
31use xmtp_events::{EventFilter, Subscription};
32use xmtp_id::associations::{AssociationState, IdentityAction, verify_updates};
33use xmtp_macro::log_event;
34use xmtp_proto::ShortHex;
35use xmtp_proto::xmtp::device_sync::content::{
36    DeviceSyncAcknowledge, PreferenceUpdates as PreferenceUpdatesProto,
37    device_sync_content::Content as ContentProto,
38};
39
40pub(super) fn log_incoming_preference_updates(
41    updates: &[xmtp_proto::xmtp::device_sync::content::PreferenceUpdate],
42) {
43    tracing::info!(update_count = updates.len(), "Incoming preference updates");
44}
45
46const RETRY_FIRST: Duration = Duration::from_secs(1);
47const RETRY_MAX: Duration = Duration::from_secs(10 * 60);
48
49/// 2 s doubling to 10 min.
50fn backoff(n: u32) -> Duration {
51    RETRY_FIRST.saturating_mul(1 << n.min(10)).min(RETRY_MAX)
52}
53
54/// A pass that is owed. Nothing is ever dropped: the work is in the database.
55#[derive(Default)]
56struct Pass {
57    due: Option<Instant>,
58    failures: u32,
59}
60
61/// The receive and outbox passes. Shared by the factory, so a worker rebuilt
62/// after a reconnect keeps their backoffs and rate-limit waits.
63#[derive(Default)]
64struct Passes {
65    inbox: Pass,
66    outbox: Pass,
67}
68
69type SharedPasses = Arc<parking_lot::Mutex<Passes>>;
70
71impl Pass {
72    /// A restart drains what is owed, but not inside a backoff.
73    fn restart(&mut self) {
74        if self.failures == 0 {
75            self.due = Some(Instant::now());
76        }
77    }
78
79    /// A wake does not cut a backoff short.
80    fn wake(&mut self) {
81        if self.failures == 0 {
82            self.due = Some(Instant::now());
83        }
84    }
85
86    fn is_due(&self) -> bool {
87        self.due.is_some_and(|at| at <= Instant::now())
88    }
89
90    // implements: SYNC-033
91    fn finish(
92        &mut self,
93        pass: &'static str,
94        result: Result<(), DeviceSyncError>,
95    ) -> Result<(), DeviceSyncError> {
96        match result {
97            Ok(()) => *self = Self::default(),
98            Err(e) if e.needs_db_reconnect() => return Err(e),
99            Err(e) => {
100                self.failures = self.failures.saturating_add(1);
101                // A retry resends the refused request, so it also waits out
102                // a rate limit's delay.
103                // implements: RATE-041
104                let delay = RetryableError::rate_limit_delay(&e).unwrap_or_default();
105                let wait = backoff(self.failures).max(delay);
106                self.due = Some(xmtp_common::time::saturating_after(Instant::now(), wait));
107                tracing::warn!(error = %e, pass, failures = self.failures, "device sync pass failed; it stays owed");
108            }
109        }
110        Ok(())
111    }
112}
113
114pub struct SyncWorker<Context> {
115    client: DeviceSyncClient<Context>,
116    subscription: Option<Arc<Subscription<InternalEvent>>>,
117    passes: SharedPasses,
118    init: OnceCell<()>,
119    metrics: Arc<WorkerMetrics<SyncMetric>>,
120}
121
122impl<Context> SyncWorker<Context>
123where
124    Context: XmtpSharedContext + 'static,
125{
126    fn new(
127        context: Context,
128        metrics: Option<Arc<WorkerMetrics<SyncMetric>>>,
129        passes: SharedPasses,
130    ) -> Self {
131        let metrics =
132            metrics.unwrap_or_else(|| Arc::new(WorkerMetrics::new(context.installation_id())));
133        let client = DeviceSyncClient::new(context, metrics.clone());
134
135        Self {
136            client,
137            subscription: None,
138            passes,
139            init: OnceCell::new(),
140            metrics,
141        }
142    }
143}
144
145struct Factory<Context> {
146    context: Context,
147    passes: SharedPasses,
148    metrics: Arc<WorkerMetrics<SyncMetric>>,
149}
150
151impl<Context> WorkerFactory for Factory<Context>
152where
153    Context: XmtpSharedContext + 'static,
154{
155    fn create(&self) -> BoxedWorker {
156        Box::new(SyncWorker::new(
157            self.context.clone(),
158            Some(self.metrics.clone()),
159            self.passes.clone(),
160        ))
161    }
162
163    fn kind(&self) -> WorkerKind {
164        WorkerKind::DeviceSync
165    }
166
167    fn event_interest(&self) -> Option<(EventFilter<InternalEvent>, Option<usize>)> {
168        let filter = EventFilter::default().with_internal(|event| {
169            matches!(
170                event,
171                InternalEvent::MessageStored { is_sync: true, .. } | InternalEvent::OutboxChanged
172            )
173        });
174        Some((filter, None))
175    }
176}
177
178impl<Context> NewWorker<Context> for SyncWorker<Context>
179where
180    Context: XmtpSharedContext + 'static,
181{
182    fn factory(context: Context) -> impl WorkerFactory + 'static {
183        Factory {
184            metrics: Arc::new(WorkerMetrics::new(context.installation_id())),
185            context,
186            passes: SharedPasses::default(),
187        }
188    }
189}
190
191#[xmtp_common::async_trait]
192impl<Context> Worker for SyncWorker<Context>
193where
194    Context: XmtpSharedContext + 'static,
195{
196    fn kind(&self) -> WorkerKind {
197        WorkerKind::DeviceSync
198    }
199
200    fn set_subscription(&mut self, subscription: Arc<Subscription<InternalEvent>>) {
201        self.subscription = Some(subscription);
202    }
203
204    fn sync_metrics(&self) -> Option<Arc<WorkerMetrics<SyncMetric>>> {
205        Some(self.metrics.clone())
206    }
207
208    async fn run_tasks(&mut self) -> WorkerResult<()> {
209        self.run().map_err(|e| Box::new(e) as Box<_>).await
210    }
211}
212
213impl<Context> SyncWorker<Context>
214where
215    Context: XmtpSharedContext + 'static,
216{
217    async fn run(&mut self) -> Result<(), DeviceSyncError> {
218        // Keep the large startup future off the worker poll stack.
219        Box::pin(self.sync_init()).await?;
220        // Receipt must outlive each sync call so remote updates can wake this worker.
221        let _receipt = IncomingCoordinator::for_context(&self.client.context)
222            .acquire(IncomingScope::DeviceSyncGroups);
223        self.metrics.increment_metric(SyncMetric::Init);
224
225        // Keep pass futures off the containing worker poll stack.
226        Box::pin(self.run_internal()).await
227    }
228
229    /// Two independent passes: a failing receive cannot block the outbox,
230    /// and the reverse. A wake marks a pass due; the tick re-reads both.
231    async fn run_internal(&mut self) -> Result<(), DeviceSyncError> {
232        use futures::StreamExt;
233        let (base, jitter) = self
234            .client
235            .context
236            .worker_interval(WorkerKind::DeviceSync, Duration::from_secs(20));
237        let mut intervals = xmtp_common::time::jittered_interval_stream(base, jitter);
238        // Poll events while native startup jitter delays the first tick.
239        let mut skip_initial_tick = cfg!(not(target_arch = "wasm32"));
240        let subscription = self
241            .subscription
242            .as_ref()
243            .expect("runner installs subscription")
244            .clone();
245        {
246            let mut passes = self.passes.lock();
247            passes.inbox.restart();
248            passes.outbox.restart();
249        }
250        loop {
251            if self.passes.lock().inbox.is_due() {
252                let result = Box::pin(self.client.process_sync_group_messages()).await;
253                self.passes.lock().inbox.finish("receive", result)?;
254            }
255            if self.passes.lock().outbox.is_due() {
256                let result = Box::pin(self.client.publish_outbox()).await;
257                self.passes.lock().outbox.finish("publish", result)?;
258            }
259            let wait = {
260                let passes = self.passes.lock();
261                [passes.inbox.due, passes.outbox.due]
262                    .into_iter()
263                    .flatten()
264                    .min()
265                    .map(|at| at.saturating_duration_since(Instant::now()))
266            };
267            tokio::select! {
268                event = subscription.next() => {
269                    let Some(event) = event else { break; };
270                    match event.internal {
271                        Some(InternalEvent::MessageStored { is_sync: true, .. }) => {
272                            self.passes.lock().inbox.wake();
273                        }
274                        Some(InternalEvent::OutboxChanged) => self.passes.lock().outbox.wake(),
275                        _ => {}
276                    }
277                }
278                _ = xmtp_common::time::sleep(wait.unwrap_or_default()), if wait.is_some() => {}
279                _ = intervals.next() => {
280                    if skip_initial_tick {
281                        skip_initial_tick = false;
282                        continue;
283                    }
284                    self.passes.lock().inbox.wake();
285                    if has_owed(&self.client.db())? {
286                        self.passes.lock().outbox.wake();
287                    }
288                },
289            }
290        }
291        Ok(())
292    }
293
294    /// Initialize the sync group when the client is registered.
295    #[instrument(level = "trace", skip_all)]
296    async fn sync_init(&mut self) -> Result<(), DeviceSyncError> {
297        let Self { init, client, .. } = &self;
298
299        init.get_or_try_init(|| async {
300            log_event!(
301                Event::DeviceSyncInitializing,
302                self.client.context.installation_id()
303            );
304
305            // The only thing that sync init really does right now is ensures that there's a sync group.
306            if client.primary_sync_group()?.is_none() {
307                log_event!(
308                    Event::DeviceSyncNoPrimarySyncGroup,
309                    self.client.context.installation_id()
310                );
311                // Sync-group creation polls membership publication below this call.
312                let sync_group = Box::pin(client.get_sync_group()).await?;
313                log_event!(
314                    Event::DeviceSyncCreatedPrimarySyncGroup,
315                    self.client.context.installation_id(),
316                    group_id = sync_group.group_id
317                );
318            }
319
320            log_event!(
321                Event::DeviceSyncInitializingFinished,
322                self.client.context.installation_id()
323            );
324
325            Ok(())
326        })
327        .await
328        .copied()
329    }
330}
331
332/// The own inbox's installation set after each own identity update, in
333/// sequence order, with the update's server time.
334struct OwnTimeline {
335    steps: Vec<(i64, HashSet<Vec<u8>>)>,
336}
337
338enum Sender {
339    Member,
340    Gone,
341    NotYetKnown,
342}
343
344impl OwnTimeline {
345    /// Load the own inbox's identity updates from the network, then build the
346    /// timeline. The load also emits own-installation events, which cycle the
347    /// root key on a revocation.
348    async fn load(context: &impl XmtpSharedContext) -> Result<Self, DeviceSyncError> {
349        let db = context.db();
350        let inbox_id = context.inbox_id();
351        load_identity_updates_for_client(context, &db, &[inbox_id]).await?;
352        let updates = db.get_identity_updates(inbox_id, None, None)?;
353        // A cached prefix state is one read.
354        let mut states = vec![];
355        for update in &updates {
356            let Some(cached) = db.read_from_cache(inbox_id, update.sequence_id)? else {
357                break;
358            };
359            states.push(AssociationState::try_from(cached).map_err(StorageError::from)?);
360        }
361        // The rest are verified in one batch, folded from the last cached
362        // state, and cached.
363        let rest = &updates[states.len()..];
364        if !rest.is_empty() {
365            let unverified = rest
366                .iter()
367                .cloned()
368                .map(IdentityExt::to_unverified)
369                .collect::<Result<Vec<_>, _>>()
370                .map_err(IdentityUpdateError::from)?;
371            let verifier = context.scw_verifier();
372            let verified = verify_updates(unverified, &verifier)
373                .await
374                .map_err(IdentityUpdateError::from)?;
375            for (stored, update) in rest.iter().zip(verified) {
376                let state = update
377                    .update_state(states.last().cloned(), update.client_timestamp_ns)
378                    .map_err(IdentityUpdateError::from)?;
379                store_association_state(context, stored.sequence_id, &state)?;
380                states.push(state);
381            }
382        }
383        let steps = updates
384            .iter()
385            .zip(states)
386            .map(|(update, state)| {
387                let installations = state.installation_ids().into_iter().collect();
388                (update.server_timestamp_ns, installations)
389            })
390            .collect();
391        Ok(Self { steps })
392    }
393
394    // implements: SYNC-030
395    fn sender_status(&self, own_inbox: &str, msg: &StoredGroupMessage) -> Sender {
396        if msg.sender_inbox_id != own_inbox {
397            // Sync groups admit only the own inbox already.
398            return Sender::Gone;
399        }
400        let id = &msg.sender_installation_id;
401        // Up to the last update, in sequence order, that the backend timed at
402        // or before the message. Server times need not grow with sequence.
403        let n = self
404            .steps
405            .iter()
406            .rposition(|(at, _)| *at <= msg.sent_at_ns)
407            .map_or(0, |last| last + 1);
408        if n > 0 && self.steps[n - 1].1.contains(id) {
409            Sender::Member
410        } else if self.steps[..n].iter().any(|(_, set)| set.contains(id)) {
411            Sender::Gone
412        } else {
413            Sender::NotYetKnown
414        }
415    }
416}
417
418impl<Context> DeviceSyncClient<Context>
419where
420    Context: XmtpSharedContext,
421{
422    /// Apply every unprocessed sync message whose sender was an own
423    /// installation at its envelope time.
424    // implements: SYNC-016, SYNC-030
425    pub(super) async fn process_sync_group_messages(&self) -> Result<(), DeviceSyncError> {
426        let mut after = None;
427        loop {
428            let page = self
429                .context
430                .db()
431                .unprocessed_sync_group_messages(after.as_ref())?;
432            let Some(last) = page.last() else {
433                return Ok(());
434            };
435            after = Some(SyncMessageKey::of(last));
436            tracing::info!("Processing {} messages.", page.len());
437            // The plaintext and its inner encoding can hold an HMAC root key.
438            // Both are wiped on every path, including skipped messages.
439            let page = with_plaintext(page);
440            // Every message in `page` was stored before this load, so the load
441            // holds every own identity update the backend timed at or before
442            // it, unless a lagging replica serves the read. A failure leaves
443            // every row Pending, and the pass backs off.
444            let own = OwnTimeline::load(&self.context).await?;
445            for (msg, plaintext) in page {
446                // Only this installation's own messages are not yet published,
447                // or failed for good; neither has an envelope time to judge.
448                match msg.delivery_status {
449                    DeliveryStatus::Unpublished => continue,
450                    DeliveryStatus::Failed => {
451                        self.context
452                            .db()
453                            .mark_device_sync_msg_as_processed(&msg.id)?;
454                        continue;
455                    }
456                    DeliveryStatus::Published => {}
457                }
458                match own.sender_status(self.context.inbox_id(), &msg) {
459                    Sender::Member => {}
460                    Sender::Gone => {
461                        tracing::warn!(
462                            group_id = msg.group_id.short_hex(),
463                            "dropping a sync message whose sender was not an own installation when it sent"
464                        );
465                        self.context
466                            .db()
467                            .mark_device_sync_msg_as_processed(&msg.id)?;
468                        continue;
469                    }
470                    // Its registration is not loaded yet: read it again later.
471                    Sender::NotYetKnown => continue,
472                }
473                if let Err(err) = self.apply_message(&msg, &plaintext) {
474                    // implements: SYNC-016 (a storage error leaves the row Pending for the next pass)
475                    // A failed message is non-fatal, but a dropped pool must stop the worker.
476                    if err.needs_db_reconnect() {
477                        return Err(err);
478                    }
479                    log_event!(
480                        Event::DeviceSyncMessageProcessingError,
481                        self.context.installation_id(),
482                        error = %err,
483                        message_id = #msg.id
484                    );
485                }
486            }
487        }
488    }
489
490    /// Apply one sync message and mark it processed, in one transaction.
491    /// Content this client does not support, such as older archive transfer
492    /// content, is marked processed.
493    pub(super) fn apply_message(
494        &self,
495        msg: &StoredGroupMessage,
496        plaintext: &[u8],
497    ) -> Result<(), DeviceSyncError> {
498        let content = decode_sync_message(plaintext);
499        let own = msg.sender_installation_id == self.installation_id().as_slice();
500        let updates = match &content {
501            Some(ReceivedContent(ContentProto::PreferenceUpdates(PreferenceUpdatesProto {
502                updates,
503            }))) => updates.as_slice(),
504            Some(ReceivedContent(ContentProto::Acknowledge(DeviceSyncAcknowledge { .. })))
505            | None => &[],
506        };
507        let msg_type = match &content {
508            Some(ReceivedContent(ContentProto::PreferenceUpdates(_))) => "PreferenceUpdates",
509            Some(ReceivedContent(ContentProto::Acknowledge(_))) => "Acknowledge",
510            None => "Unsupported",
511        };
512        log_event!(
513            Event::DeviceSyncProcessingMessages,
514            self.context.installation_id(),
515            msg_type,
516            external = !own,
517            message_id = #msg.id,
518            group_id = msg.group_id
519        );
520        if !own {
521            log_incoming_preference_updates(updates);
522        }
523        // implements: PROC-036
524        tracing::info!(update_count = updates.len(), "storing preference updates");
525        crate::state_tx::state_write_with_events(
526            self.context.mls_storage(),
527            self.context.events(),
528            |tx, events| {
529                // Read in this write, so a sync Welcome commits wholly before
530                // or after it.
531                let selected = selected_sync_group_in(tx, self.context.inbox_id())?;
532                let storage = tx.storage();
533                let db = storage.db();
534                let applied =
535                    apply_preference_updates(&db, self.context.inbox_id(), msg, updates, selected)?;
536                emit_consent_changes(&self.context, events, &db, &applied.consent)?;
537                if let Some(marked) = applied.root_key {
538                    emit_root_key_changed(events, &db, marked)?;
539                }
540                db.mark_device_sync_msg_as_processed(&msg.id)?;
541                Ok::<_, crate::groups::GroupError>(Continue(()))
542            },
543        )?;
544        for update in updates.iter().filter_map(|update| update.update.as_ref()) {
545            if !own {
546                self.metrics.increment_metric(match update {
547                    super::PreferenceUpdateProto::Consent(_) => SyncMetric::ConsentReceived,
548                    super::PreferenceUpdateProto::Hmac(_) => SyncMetric::HmacReceived,
549                });
550            }
551        }
552        Ok(())
553    }
554}
555
556#[cfg(test)]
557mod startup_tests {
558    use super::*;
559    use crate::prelude::*;
560    use crate::tester;
561
562    fn idle_worker<C: XmtpSharedContext + 'static>(context: C) -> SyncWorker<C> {
563        SyncWorker::new(context, None, SharedPasses::default())
564    }
565
566    /// A failed pass stays owed and waits out its backoff; a wake does not
567    /// cut the backoff short; a success clears it.
568    // verifies: SYNC-033
569    #[xmtp_common::test(unwrap_try = true)]
570    fn pass_backs_off_and_stays_owed() {
571        let mut pass = Pass::default();
572        pass.restart();
573        assert!(pass.is_due());
574        pass.finish("test", Err(DeviceSyncError::OutboxPageUnpublished))?;
575        assert_eq!(pass.failures, 1);
576        assert!(pass.due.is_some(), "a failed pass stays owed");
577        assert!(!pass.is_due());
578        pass.wake();
579        assert!(!pass.is_due(), "a wake does not cut a backoff short");
580        pass.finish("test", Err(DeviceSyncError::OutboxPageUnpublished))?;
581        assert_eq!(pass.failures, 2);
582        assert!(backoff(2) > backoff(1));
583        pass.finish("test", Ok(()))?;
584        assert_eq!((pass.due, pass.failures), (None, 0));
585        pass.wake();
586        assert!(pass.is_due());
587    }
588
589    /// A pass refused by a rate limit runs again only after the delay: a wake
590    /// or a tick does not cut it short, and neither does a worker rebuilt
591    /// after a reconnect, which shares the passes through the factory. A
592    /// refused publish and a refused own-identity read both wait.
593    // verifies: RATE-041
594    #[xmtp_common::test(unwrap_try = true)]
595    async fn a_rate_limited_pass_waits_out_its_delay() {
596        tester!(alix, disable_workers);
597        let delay = Duration::from_secs(300);
598        let refused = |method| {
599            xmtp_api::ApiError::Api(xmtp_proto::api::NetworkError::new(
600                xmtp_api_grpc::error::GrpcError::Status(xmtp_api_grpc::error::rate_limited_status(
601                    method, delay,
602                )),
603            ))
604        };
605        let publish =
606            DeviceSyncError::Group(crate::groups::GroupError::WrappedApi(refused("publish")));
607        let identity = DeviceSyncError::IdentityUpdate(
608            crate::identity_updates::IdentityUpdateError::Api(refused("get_identity_updates")),
609        );
610        for error in [&publish, &identity] {
611            assert_eq!(RetryableError::rate_limit_delay(error), Some(delay));
612        }
613        let factory = Factory {
614            context: alix.context.clone(),
615            passes: SharedPasses::default(),
616            metrics: Arc::new(WorkerMetrics::new(alix.context.installation_id())),
617        };
618        let before = Instant::now();
619        {
620            let mut passes = factory.passes.lock();
621            passes.outbox.finish("publish", Err(publish))?;
622            passes.inbox.finish("receive", Err(identity))?;
623        }
624        let _rebuilt = factory.create();
625        assert_eq!(
626            Arc::strong_count(&factory.passes),
627            2,
628            "a rebuilt worker shares the passes"
629        );
630        let mut guard = factory.passes.lock();
631        let passes = &mut *guard;
632        for pass in [&mut passes.outbox, &mut passes.inbox] {
633            pass.wake();
634            pass.restart();
635            assert!(!pass.is_due());
636            assert!(pass.due.unwrap() >= before + delay);
637        }
638    }
639
640    /// The sender of a sync message is judged at its envelope time against
641    /// the own identity history.
642    // verifies: SYNC-030
643    #[xmtp_common::test]
644    fn sender_is_judged_at_the_envelope_time() {
645        let set = |ids: &[u8]| ids.iter().map(|id| vec![*id]).collect::<HashSet<_>>();
646        let timeline = OwnTimeline {
647            steps: vec![(10, set(&[1])), (20, set(&[1, 2])), (30, set(&[2]))],
648        };
649        let message = |inbox: &str, installation: u8, sent_at_ns| StoredGroupMessage {
650            sender_inbox_id: inbox.into(),
651            sender_installation_id: vec![installation],
652            sent_at_ns,
653            ..crate::test::mock::generate_stored_msg(
654                xmtp_proto::types::Cursor(1),
655                xmtp_proto::types::GroupId::from([1; 16]),
656            )
657        };
658        // The backend's clock stepped back: 3 is revoked at 60, then an
659        // update is timed at 50.
660        let stepped_back = OwnTimeline {
661            steps: vec![(40, set(&[3])), (60, set(&[4])), (50, set(&[4]))],
662        };
663        for (at, expected) in [(45, "member"), (55, "gone")] {
664            let status = match stepped_back.sender_status("me", &message("me", 3, at)) {
665                Sender::Member => "member",
666                Sender::Gone => "gone",
667                Sender::NotYetKnown => "not yet known",
668            };
669            assert_eq!(status, expected, "3 at {at} after the step back");
670        }
671        for (inbox, installation, at, expected) in [
672            ("me", 1, 15, "member"),
673            // Sent before its revocation at 30: applied, whenever it is processed.
674            ("me", 1, 29, "member"),
675            ("me", 1, 30, "gone"),
676            ("me", 2, 15, "not yet known"),
677            ("me", 2, 20, "member"),
678            ("me", 3, 40, "not yet known"),
679            ("other", 1, 15, "gone"),
680        ] {
681            let status = match timeline.sender_status("me", &message(inbox, installation, at)) {
682                Sender::Member => "member",
683                Sender::Gone => "gone",
684                Sender::NotYetKnown => "not yet known",
685            };
686            assert_eq!(status, expected, "{inbox} {installation} at {at}");
687        }
688    }
689
690    /// A timeline built from the cache and one verified pass gives each
691    /// prefix the state the resolver gives it, and caches each one.
692    // verifies: SYNC-030
693    #[xmtp_common::test(unwrap_try = true)]
694    async fn timeline_states_match_each_prefix() {
695        use diesel::prelude::*;
696        use xmtp_db::{ConnectionExt, schema::association_state::dsl};
697        tester!(alix1, disable_workers);
698        tester!(alix2, from: alix1, disable_workers);
699        tester!(alix3, from: alix1, disable_workers);
700        let db = alix1.context.db();
701        let inbox_id = alix1.inbox_id();
702        load_identity_updates_for_client(&alix1.context, &db, &[inbox_id]).await?;
703        let updates = db.get_identity_updates(inbox_id, None, None)?;
704        assert!(updates.len() >= 3);
705        let mut expected = vec![];
706        for update in &updates {
707            let state = crate::identity_updates::IdentityUpdates::new(&alix1.context)
708                .get_association_state(&db, inbox_id, Some(update.sequence_id))
709                .await?;
710            let installations: HashSet<_> = state.installation_ids().into_iter().collect();
711            expected.push((update.server_timestamp_ns, installations));
712        }
713        // Only the first prefix stays cached, so the fold starts from it.
714        db.raw_query(|conn| {
715            diesel::delete(
716                dsl::association_state.filter(dsl::sequence_id.gt(updates[0].sequence_id)),
717            )
718            .execute(conn)
719        })?;
720        let timeline = OwnTimeline::load(&alix1.context).await?;
721        assert_eq!(timeline.steps, expected);
722        for update in &updates {
723            assert!(db.read_from_cache(inbox_id, update.sequence_id)?.is_some());
724        }
725        drop((alix2, alix3));
726    }
727
728    /// Closing the subscription stops the worker after its owed passes.
729    #[xmtp_common::test(unwrap_try = true)]
730    async fn closing_subscription_stops_worker() {
731        tester!(alix, disable_workers);
732        let mut worker = idle_worker(alix.context.clone());
733        let (filter, depth) = SyncWorker::factory(alix.context.clone())
734            .event_interest()
735            .unwrap();
736        let subscription = Arc::new(alix.context.events().subscribe(filter, depth));
737        worker.set_subscription(subscription.clone());
738        subscription.close();
739        xmtp_common::time::timeout(Duration::from_secs(60), worker.run_internal()).await??;
740        alix.close().await?;
741    }
742
743    /// The publish span carries no root key, also when a root key is owed.
744    #[cfg(not(target_arch = "wasm32"))]
745    #[xmtp_common::test(unwrap_try = true)]
746    async fn publish_span_omits_hmac_key() {
747        tester!(alix, disable_workers);
748        let db = alix.context.db();
749        let (key, _) = xmtp_db::user_preferences::ensure_root_key(&db)?;
750        let key_debug = format!("{:?}", key.as_slice());
751        let client = alix.device_sync_client();
752        let (result, spans) =
753            xmtp_logging::test_logging::with_trace_layer_async(true, client.publish_outbox()).await;
754        result?;
755        let span = spans
756            .iter()
757            .find(|span| span.name == "publish_outbox")
758            .expect("the outbox exports its span");
759        let fields = format!("{:?}", span.attributes);
760        assert!(fields.contains("publish_outbox"));
761        assert!(
762            !fields.contains(&key_debug),
763            "the publish span must not export the HMAC root key"
764        );
765        assert!(!has_owed(&db)?, "the root key was published");
766        alix.close().await?;
767    }
768}
769
770#[cfg(test)]
771mod tests {
772    use super::SyncWorker;
773    use crate::core::events::InternalEvent;
774    use crate::test::mock::{NewMockContext, context};
775    use crate::worker::{NewWorker, WorkerFactory, WorkerKind};
776    use xmtp_events::{EventBus, EventWriter};
777
778    /// The sync worker's event interest, as the runner subscribes with it.
779    fn worker_event_filter(
780        kind: WorkerKind,
781    ) -> Option<(xmtp_events::EventFilter<InternalEvent>, Option<usize>)> {
782        let factory = SyncWorker::<NewMockContext>::factory(context());
783        assert_eq!(factory.kind(), kind);
784        factory.event_interest()
785    }
786
787    /// The outbox and the sync messages are stored state; the worker wakes
788    /// only for them, and never drops a wake while busy.
789    // verifies: SYNC-029
790    #[xmtp_common::test(unwrap_try = true)]
791    async fn device_sync_wakes_for_sync_messages_and_the_outbox() {
792        let (filter, depth) = worker_event_filter(WorkerKind::DeviceSync).unwrap();
793        assert_eq!(depth, None);
794        let bus = EventBus::new();
795        let subscription = bus.subscribe(filter, depth);
796        let group_id = xmtp_proto::types::GroupId::from([9; 16]);
797        let stored = |is_sync| InternalEvent::MessageStored {
798            group_id,
799            message_id: vec![],
800            expires_at_ns: None,
801            is_sync,
802        };
803        for event in [
804            InternalEvent::PreferencesChanged { updates: vec![] },
805            InternalEvent::GroupJoined {
806                group_id,
807                is_sync: true,
808            },
809            stored(false),
810        ] {
811            bus.emit(None, Some(event));
812        }
813        assert!(subscription.drain().is_empty());
814        for _ in 0..1_000 {
815            bus.emit(None, Some(InternalEvent::OutboxChanged));
816        }
817        bus.emit(None, Some(stored(true)));
818        assert_eq!(subscription.drain().len(), 1_001);
819    }
820}
xmtp_mls/worker/device_sync/worker.rs

xmtp_mls/worker/device_sync/
worker.rs
