xmtp_mls/
context.rs

1#[cfg(test)]
2use crate::GroupCommitLock;
3use crate::builder::{DeviceSyncMode, ForkRecoveryOpts};
4#[cfg(any(test, target_arch = "wasm32"))]
5mod message_clock;
6use crate::client::DeviceSync;
7use crate::groups::change_callbacks::UnstableChangeCallbacks;
8use crate::server_configuration::ServerConfigurationHandle;
9use crate::subscriptions::internal::InternalEvent;
10use crate::utils::VersionInfo;
11use crate::worker::device_sync::worker::SyncMetric;
12use crate::worker::metrics::WorkerMetrics;
13use crate::worker::tasks::TaskWorkerChannels;
14use crate::worker::{DynMetrics, MetricsCasting, WorkerConfig, WorkerKind};
15use crate::{
16    identity::{Identity, IdentityError},
17    mutex_registry::MutexRegistry,
18};
19use parking_lot::Mutex;
20use std::collections::HashMap;
21use std::sync::Arc;
22use std::sync::atomic::{AtomicBool, Ordering};
23use tokio_util::sync::CancellationToken;
24use xmtp_api::{ApiClientWrapper, XmtpApi};
25use xmtp_common::{MaybeSend, MaybeSync};
26use xmtp_db::XmtpDb;
27use xmtp_db::XmtpMlsStorageProvider;
28use xmtp_db::xmtp_openmls_provider::XmtpOpenMlsProviderRef;
29use xmtp_events::EventBus;
30use xmtp_events::EventWriter;
31use xmtp_id::scw_verifier::SmartContractSignatureVerifier;
32use xmtp_id::{InboxIdRef, associations::builder::SignatureRequest};
33use xmtp_proto::types::InstallationId;
34
35#[cfg(test)]
36pub(crate) static FAIL_NEXT_DELIVERY_RELEASE: Mutex<Option<Vec<u8>>> = Mutex::new(None);
37
38#[cfg(any(test, feature = "test-utils"))]
39use crate::worker::device_sync::DeviceSyncClient;
40
41/// Counts the running host calls that can use the database.
42///
43/// `Client::close` cancels the client, then closes this gate and waits until
44/// no call runs. Only after that does it stop the workers and disconnect the
45/// database, so a call such as `reconnect_db` cannot start work that outlives
46/// close. A call that is inside the gate must not wait for `Client::close` of
47/// its own client, or for a worker, because close waits for that call.
48/// Only request calls enter the gate. A stream or a reader must stop on
49/// cancellation, or close does not finish.
50#[derive(Default)]
51pub struct ForegroundCalls {
52    state: Mutex<ForegroundCallState>,
53    idle: tokio::sync::Notify,
54}
55
56#[derive(Default)]
57struct ForegroundCallState {
58    closed: bool,
59    running: usize,
60}
61
62/// One running call. Drop it when the call no longer uses the database.
63#[must_use = "the call leaves the gate when this guard drops"]
64pub struct ForegroundCall {
65    calls: Arc<ForegroundCalls>,
66}
67
68impl ForegroundCalls {
69    /// Start a call, or return `None` after close has begun.
70    pub fn enter(self: &Arc<Self>) -> Option<ForegroundCall> {
71        let mut state = self.state.lock();
72        if state.closed {
73            return None;
74        }
75        state.running += 1;
76        Some(ForegroundCall {
77            calls: self.clone(),
78        })
79    }
80
81    /// Refuse new calls, then wait until the running calls finish.
82    pub async fn close_and_wait(&self) {
83        self.state.lock().closed = true;
84        loop {
85            let idle = self.idle.notified();
86            futures::pin_mut!(idle);
87            idle.as_mut().enable();
88            if self.state.lock().running == 0 {
89                return;
90            }
91            idle.await;
92        }
93    }
94}
95
96impl Drop for ForegroundCall {
97    fn drop(&mut self) {
98        let mut state = self.calls.state.lock();
99        state.running -= 1;
100        if state.running == 0 {
101            self.calls.idle.notify_waiters();
102        }
103    }
104}
105
106/// The local context a XMTP MLS needs to function:
107/// - Sqlite Database
108/// - Identity for the User
109pub struct XmtpMlsLocalContext<ApiClient, Db, S> {
110    pub(crate) attachments: Arc<crate::attachments::AttachmentRuntime>,
111    /// XMTP Identity
112    pub(crate) identity: Identity,
113    /// The XMTP Api Client
114    pub(crate) api_client: ApiClientWrapper<ApiClient>,
115    /// XMTP Local Storage
116    pub(crate) store: Db,
117    pub(crate) mls_storage: S,
118    pub(crate) mutexes: MutexRegistry,
119    #[cfg(test)]
120    pub(crate) mls_commit_lock: Arc<GroupCommitLock>,
121    pub(crate) version_info: VersionInfo,
122    /// What this deployment published about itself, resolved once at build
123    /// plus the blocked connection a refresh may set.
124    pub(crate) server_configuration: ServerConfigurationHandle,
125    pub(crate) events: EventBus<InternalEvent>,
126    pub(crate) public_event_writer: Arc<dyn EventWriter<()>>,
127    /// Set by a new registration and cleared after its visibility event.
128    pub(crate) registration_event_pending: Arc<AtomicBool>,
129    pub(crate) delivery_owner: Arc<Mutex<Option<xmtp_db::delivery::DeliveryOwner>>>,
130    pub(crate) scw_verifier: Arc<Box<dyn SmartContractSignatureVerifier>>,
131    pub(crate) device_sync: DeviceSync,
132    pub(crate) fork_recovery_opts: ForkRecoveryOpts,
133    /// Unstable: SDK-registered notifications for group-state changes. Empty
134    /// unless the host opted in at build time.
135    pub(crate) change_callbacks: UnstableChangeCallbacks,
136    pub(crate) incoming_runtime: Arc<crate::subscriptions::incoming::IncomingRuntime>,
137    pub(crate) identity_resolutions: Arc<crate::identity_updates::IdentityResolutionRegistry>,
138    pub(crate) worker_config: WorkerConfig,
139    // pub(crate) workers: Arc<WorkerRunner>,
140    pub(crate) worker_metrics: Arc<Mutex<HashMap<WorkerKind, DynMetrics>>>,
141    pub(crate) task_channels: TaskWorkerChannels,
142    pub(crate) cancellation_token: CancellationToken,
143    // Set only after a successful `Client::close` (workers stopped + DB
144    // disconnected). The cancellation token tracks "shutdown initiated";
145    // this tracks "shutdown completed cleanly" — distinct semantics so
146    // a `disconnect()` failure mid-`close` doesn't silently short-circuit
147    // future retries.
148    pub(crate) shutdown_complete: Arc<AtomicBool>,
149    pub(crate) foreground_calls: Arc<ForegroundCalls>,
150    #[cfg(any(test, target_arch = "wasm32"))]
151    pub(crate) message_clock: Arc<message_clock::MessageClock>,
152}
153
154impl<ApiClient, Db, S> XmtpMlsLocalContext<ApiClient, Db, S>
155where
156    Db: XmtpDb + 'static,
157    ApiClient: XmtpApi + 'static,
158    S: XmtpMlsStorageProvider + 'static,
159{
160    /// get a reference to the monolithic Database object where
161    /// higher-level queries are defined
162    pub fn db(&self) -> Db::DbQuery {
163        self.store.db()
164    }
165
166    /// Creates a new MLS Provider
167    pub fn mls_provider(&'_ self) -> XmtpOpenMlsProviderRef<'_, S> {
168        XmtpOpenMlsProviderRef::new(&self.mls_storage)
169    }
170
171    pub fn store(&self) -> &Db {
172        &self.store
173    }
174
175    pub fn scw_verifier(&self) -> &Arc<Box<dyn SmartContractSignatureVerifier>> {
176        &self.scw_verifier
177    }
178
179    pub fn device_sync_worker_enabled(&self) -> bool {
180        !matches!(self.device_sync.mode, DeviceSyncMode::Disabled)
181    }
182
183    /// Reconstructs the DeviceSyncClient from the context
184    /// used in tests
185    #[cfg(any(test, feature = "test-utils"))]
186    pub fn device_sync_client(
187        self: &Arc<XmtpMlsLocalContext<ApiClient, Db, S>>,
188    ) -> DeviceSyncClient<Arc<Self>> {
189        let metrics = self.sync_metrics();
190        DeviceSyncClient::new(
191            Arc::clone(self),
192            metrics.unwrap_or(Arc::new(WorkerMetrics::new(self.installation_id()))),
193        )
194    }
195}
196
197impl<ApiClient, Db, S> XmtpMlsLocalContext<ApiClient, Db, S> {
198    pub fn replace_mls_store<S2>(self, mls_store: S2) -> XmtpMlsLocalContext<ApiClient, Db, S2> {
199        XmtpMlsLocalContext::<ApiClient, Db, S2> {
200            attachments: self.attachments,
201            identity: self.identity,
202            api_client: self.api_client,
203            store: self.store,
204            mls_storage: mls_store,
205            mutexes: self.mutexes,
206            #[cfg(test)]
207            mls_commit_lock: self.mls_commit_lock,
208            version_info: self.version_info,
209            server_configuration: self.server_configuration,
210            events: self.events,
211            public_event_writer: self.public_event_writer,
212            registration_event_pending: self.registration_event_pending,
213            delivery_owner: self.delivery_owner,
214            scw_verifier: self.scw_verifier,
215            device_sync: self.device_sync,
216            fork_recovery_opts: self.fork_recovery_opts,
217            change_callbacks: self.change_callbacks,
218            incoming_runtime: self.incoming_runtime,
219            identity_resolutions: self.identity_resolutions,
220            worker_config: self.worker_config,
221            worker_metrics: self.worker_metrics,
222            task_channels: self.task_channels,
223            cancellation_token: self.cancellation_token,
224            shutdown_complete: self.shutdown_complete,
225            foreground_calls: self.foreground_calls,
226            #[cfg(any(test, target_arch = "wasm32"))]
227            message_clock: self.message_clock,
228        }
229    }
230}
231
232impl<ApiClient, Db, S> XmtpMlsLocalContext<ApiClient, Db, S> {
233    /// The installation public key is the primary identifier for an installation
234    pub fn installation_public_key(&self) -> InstallationId {
235        (*self.identity.installation_keys.public_bytes()).into()
236    }
237
238    /// The installation public key is the primary identifier for an installation
239    pub fn installation_id(&self) -> InstallationId {
240        self.identity.installation_id()
241    }
242
243    /// Get the account address of the blockchain account associated with this client
244    pub fn inbox_id(&self) -> InboxIdRef<'_> {
245        self.identity.inbox_id()
246    }
247
248    /// Integrators should always check the `signature_request` return value of this function before calling `register_identity`.
249    /// If `signature_request` returns `None`, then the wallet signature is not required and `register_identity` can be called with None as an argument.
250    pub fn signature_request(&self) -> Option<SignatureRequest> {
251        self.identity.signature_request()
252    }
253
254    pub fn sign_with_public_context(
255        &self,
256        text: impl AsRef<str>,
257    ) -> Result<Vec<u8>, IdentityError> {
258        self.identity.sign_with_public_context(text)
259    }
260
261    #[cfg(test)]
262    pub fn mls_commit_lock(&self) -> &Arc<GroupCommitLock> {
263        &self.mls_commit_lock
264    }
265
266    pub fn sync_metrics(&self) -> Option<Arc<WorkerMetrics<SyncMetric>>> {
267        self.worker_metrics
268            .lock()
269            .get(&WorkerKind::DeviceSync)?
270            .as_sync_metrics()
271    }
272
273    pub fn cancellation_token(&self) -> &CancellationToken {
274        &self.cancellation_token
275    }
276
277    pub fn shutdown_complete(&self) -> bool {
278        self.shutdown_complete.load(Ordering::Acquire)
279    }
280
281    pub fn mark_shutdown_complete(&self) {
282        self.shutdown_complete.store(true, Ordering::Release);
283    }
284}
285
286pub trait XmtpSharedContext
287where
288    Self: MaybeSend + MaybeSync + Sized + Clone,
289{
290    type Db: XmtpDb;
291    type ApiClient: XmtpApi;
292    type MlsStorage: XmtpMlsStorageProvider;
293    /// Owned, cloneable handle for background work; it cannot borrow short-lived resources.
294    type ContextReference: XmtpSharedContext<Db = Self::Db, ApiClient = Self::ApiClient, MlsStorage = Self::MlsStorage>
295        + 'static;
296
297    /// Return the handle to clone when work can outlive this context borrow.
298    fn context_ref(&self) -> &Self::ContextReference;
299    fn db(&self) -> <Self::Db as XmtpDb>::DbQuery;
300    fn attachment_runtime(&self) -> &Arc<crate::attachments::AttachmentRuntime>;
301    fn api(&self) -> &ApiClientWrapper<Self::ApiClient>;
302    fn scw_verifier(&self) -> Arc<Box<dyn SmartContractSignatureVerifier>>;
303
304    fn device_sync(&self) -> &DeviceSync;
305
306    fn device_sync_worker_enabled(&self) -> bool {
307        !matches!(self.device_sync().mode, DeviceSyncMode::Disabled)
308    }
309
310    fn fork_recovery_opts(&self) -> &ForkRecoveryOpts;
311
312    fn worker_config(&self) -> &WorkerConfig;
313
314    /// Resolve `(base, jitter)` for a worker from the configured
315    /// [`WorkerConfig`], falling back to the worker's compiled-in const.
316    fn worker_interval(
317        &self,
318        kind: WorkerKind,
319        const_default: std::time::Duration,
320    ) -> (std::time::Duration, std::time::Duration) {
321        self.worker_config().interval(kind, const_default)
322    }
323
324    /// Creates a new MLS Provider
325    fn mls_provider(&'_ self) -> XmtpOpenMlsProviderRef<'_, Self::MlsStorage> {
326        XmtpOpenMlsProviderRef::new(self.mls_storage())
327    }
328
329    fn mls_storage(&self) -> &Self::MlsStorage;
330    fn identity(&self) -> &Identity;
331
332    fn signature_request(&self) -> Option<SignatureRequest> {
333        self.identity().signature_request()
334    }
335
336    fn inbox_id(&self) -> InboxIdRef<'_> {
337        self.identity().inbox_id()
338    }
339
340    fn installation_id(&self) -> InstallationId {
341        (*self.identity().installation_keys.public_bytes()).into()
342    }
343
344    fn version_info(&self) -> &VersionInfo;
345    /// The configuration snapshot every consumer reads.
346    fn server_configuration(&self) -> &ServerConfigurationHandle;
347    fn events(&self) -> &EventBus<InternalEvent>;
348    fn registration_event_pending(&self) -> &AtomicBool;
349    fn public_event_writer(&self) -> &Arc<dyn EventWriter<()>>;
350    /// This context's default-consumer token; the database is the ownership authority.
351    fn delivery_owner(&self) -> &Mutex<Option<xmtp_db::delivery::DeliveryOwner>>;
352
353    /// Release this context's message consumer before disconnecting its database.
354    fn close_message_delivery(&self) -> Result<(), xmtp_db::StorageError> {
355        use xmtp_db::delivery::QueryDelivery;
356        #[cfg(test)]
357        {
358            let mut fail = FAIL_NEXT_DELIVERY_RELEASE.lock();
359            if fail.as_deref() == Some(self.installation_id().to_vec().as_slice()) {
360                *fail = None;
361                return Err(xmtp_db::StorageError::DbDeserialize);
362            }
363        }
364        let mut registered = self.delivery_owner().lock();
365        if let Some(owner) = *registered {
366            self.db().release_delivery_owner(owner)?;
367            *registered = None;
368        }
369        Ok(())
370    }
371    fn task_channels(&self) -> &TaskWorkerChannels;
372    /// Unstable: the host's registered group-change callbacks.
373    fn change_callbacks(&self) -> &UnstableChangeCallbacks;
374    /// Shared incoming runtime. Its limits and transport are internal client policy.
375    fn incoming_runtime(&self) -> &crate::subscriptions::incoming::IncomingRuntime;
376    /// Coalesces exact identity lookups without treating a newer snapshot as the requested one.
377    fn identity_resolution_registry(&self) -> &crate::identity_updates::IdentityResolutionRegistry;
378    fn sync_metrics(&self) -> Option<Arc<WorkerMetrics<SyncMetric>>>;
379    #[cfg(test)]
380    fn mls_commit_lock(&self) -> &Arc<GroupCommitLock>;
381    fn mutexes(&self) -> &MutexRegistry;
382    fn cancellation_token(&self) -> &CancellationToken;
383
384    /// Returns `true` once `Client::close` has been called on this context.
385    fn is_closed(&self) -> bool {
386        self.cancellation_token().is_cancelled()
387    }
388
389    /// Returns `true` only after `Client::close` has fully torn the client
390    /// down (workers drained + DB disconnected). Distinct from [`Self::is_closed`]
391    /// which fires the moment shutdown begins — this guards `close`'s
392    /// idempotency check so a mid-shutdown failure stays retryable.
393    fn shutdown_complete(&self) -> bool;
394
395    fn mark_shutdown_complete(&self);
396
397    /// Calls that `Client::close` waits for before it disconnects the database.
398    fn foreground_calls(&self) -> &Arc<ForegroundCalls>;
399
400    /// Timestamp for a newly accepted local message. WASM sends are monotonic.
401    fn message_timestamp_ns(&self) -> i64;
402}
403
404impl<XApiClient, XDb, XMls> XmtpSharedContext for Arc<XmtpMlsLocalContext<XApiClient, XDb, XMls>>
405where
406    XApiClient: XmtpApi + 'static,
407    XDb: XmtpDb + 'static,
408    XMls: XmtpMlsStorageProvider + 'static,
409{
410    type Db = XDb;
411    type ApiClient = XApiClient;
412    type MlsStorage = XMls;
413    type ContextReference = Self;
414
415    fn context_ref(&self) -> &Self::ContextReference {
416        self
417    }
418
419    fn db(&self) -> <Self::Db as XmtpDb>::DbQuery {
420        self.store.db()
421    }
422
423    fn attachment_runtime(&self) -> &Arc<crate::attachments::AttachmentRuntime> {
424        &self.attachments
425    }
426
427    fn api(&self) -> &ApiClientWrapper<Self::ApiClient> {
428        &self.api_client
429    }
430
431    fn scw_verifier(&self) -> Arc<Box<dyn SmartContractSignatureVerifier>> {
432        self.scw_verifier.clone()
433    }
434
435    fn device_sync(&self) -> &DeviceSync {
436        &self.device_sync
437    }
438
439    fn fork_recovery_opts(&self) -> &ForkRecoveryOpts {
440        &self.fork_recovery_opts
441    }
442
443    fn worker_config(&self) -> &WorkerConfig {
444        &self.worker_config
445    }
446
447    /// a reference to the MLS Storage Type
448    /// This can be related to 'db()' but may also be separate
449    fn mls_storage(&self) -> &Self::MlsStorage {
450        &self.mls_storage
451    }
452
453    fn identity(&self) -> &Identity {
454        &self.identity
455    }
456
457    fn version_info(&self) -> &VersionInfo {
458        &self.version_info
459    }
460
461    fn server_configuration(&self) -> &ServerConfigurationHandle {
462        &self.server_configuration
463    }
464
465    fn events(&self) -> &EventBus<InternalEvent> {
466        &self.events
467    }
468
469    fn registration_event_pending(&self) -> &AtomicBool {
470        &self.registration_event_pending
471    }
472
473    fn public_event_writer(&self) -> &Arc<dyn EventWriter<()>> {
474        &self.public_event_writer
475    }
476
477    fn delivery_owner(&self) -> &Mutex<Option<xmtp_db::delivery::DeliveryOwner>> {
478        &self.delivery_owner
479    }
480
481    #[cfg(test)]
482    fn mls_commit_lock(&self) -> &Arc<GroupCommitLock> {
483        &self.mls_commit_lock
484    }
485
486    fn task_channels(&self) -> &TaskWorkerChannels {
487        &self.task_channels
488    }
489
490    fn change_callbacks(&self) -> &UnstableChangeCallbacks {
491        &self.change_callbacks
492    }
493
494    fn incoming_runtime(&self) -> &crate::subscriptions::incoming::IncomingRuntime {
495        &self.incoming_runtime
496    }
497
498    fn identity_resolution_registry(&self) -> &crate::identity_updates::IdentityResolutionRegistry {
499        &self.identity_resolutions
500    }
501
502    fn sync_metrics(&self) -> Option<Arc<WorkerMetrics<SyncMetric>>> {
503        self.worker_metrics
504            .lock()
505            .get(&WorkerKind::DeviceSync)?
506            .as_sync_metrics()
507    }
508
509    fn mutexes(&self) -> &MutexRegistry {
510        &self.mutexes
511    }
512
513    fn cancellation_token(&self) -> &CancellationToken {
514        &self.cancellation_token
515    }
516
517    fn shutdown_complete(&self) -> bool {
518        self.shutdown_complete.load(Ordering::Acquire)
519    }
520
521    fn mark_shutdown_complete(&self) {
522        self.shutdown_complete.store(true, Ordering::Release);
523    }
524
525    fn foreground_calls(&self) -> &Arc<ForegroundCalls> {
526        &self.foreground_calls
527    }
528
529    fn message_timestamp_ns(&self) -> i64 {
530        #[cfg(target_arch = "wasm32")]
531        {
532            self.message_clock.now_ns()
533        }
534
535        #[cfg(not(target_arch = "wasm32"))]
536        {
537            #[cfg(test)]
538            if self.message_clock.is_frozen() {
539                return self.message_clock.now_ns();
540            }
541            xmtp_common::time::now_ns()
542        }
543    }
544}
545
546impl<T> XmtpSharedContext for &T
547where
548    T: XmtpSharedContext,
549{
550    type Db = <T as XmtpSharedContext>::Db;
551    type ApiClient = <T as XmtpSharedContext>::ApiClient;
552    type MlsStorage = <T as XmtpSharedContext>::MlsStorage;
553    type ContextReference = <T as XmtpSharedContext>::ContextReference;
554
555    fn context_ref(&self) -> &Self::ContextReference {
556        <T as XmtpSharedContext>::context_ref(self)
557    }
558
559    fn db(&self) -> <Self::Db as XmtpDb>::DbQuery {
560        <T as XmtpSharedContext>::db(self)
561    }
562
563    fn attachment_runtime(&self) -> &Arc<crate::attachments::AttachmentRuntime> {
564        <T as XmtpSharedContext>::attachment_runtime(self)
565    }
566
567    fn api(&self) -> &ApiClientWrapper<Self::ApiClient> {
568        <T as XmtpSharedContext>::api(self)
569    }
570
571    fn scw_verifier(&self) -> Arc<Box<dyn SmartContractSignatureVerifier>> {
572        <T as XmtpSharedContext>::scw_verifier(self)
573    }
574
575    fn device_sync(&self) -> &DeviceSync {
576        <T as XmtpSharedContext>::device_sync(self)
577    }
578
579    fn device_sync_worker_enabled(&self) -> bool {
580        <T as XmtpSharedContext>::device_sync_worker_enabled(self)
581    }
582
583    fn fork_recovery_opts(&self) -> &ForkRecoveryOpts {
584        <T as XmtpSharedContext>::fork_recovery_opts(self)
585    }
586
587    fn worker_config(&self) -> &WorkerConfig {
588        <T as XmtpSharedContext>::worker_config(self)
589    }
590
591    fn mls_storage(&self) -> &Self::MlsStorage {
592        <T as XmtpSharedContext>::mls_storage(self)
593    }
594
595    fn identity(&self) -> &Identity {
596        <T as XmtpSharedContext>::identity(self)
597    }
598
599    fn version_info(&self) -> &VersionInfo {
600        <T as XmtpSharedContext>::version_info(self)
601    }
602
603    fn server_configuration(&self) -> &ServerConfigurationHandle {
604        <T as XmtpSharedContext>::server_configuration(self)
605    }
606
607    fn events(&self) -> &EventBus<InternalEvent> {
608        <T as XmtpSharedContext>::events(self)
609    }
610
611    fn registration_event_pending(&self) -> &AtomicBool {
612        <T as XmtpSharedContext>::registration_event_pending(self)
613    }
614
615    fn public_event_writer(&self) -> &Arc<dyn EventWriter<()>> {
616        <T as XmtpSharedContext>::public_event_writer(self)
617    }
618
619    fn delivery_owner(&self) -> &Mutex<Option<xmtp_db::delivery::DeliveryOwner>> {
620        <T as XmtpSharedContext>::delivery_owner(self)
621    }
622
623    #[cfg(test)]
624    fn mls_commit_lock(&self) -> &Arc<GroupCommitLock> {
625        <T as XmtpSharedContext>::mls_commit_lock(self)
626    }
627
628    fn task_channels(&self) -> &TaskWorkerChannels {
629        <T as XmtpSharedContext>::task_channels(self)
630    }
631
632    fn change_callbacks(&self) -> &UnstableChangeCallbacks {
633        <T as XmtpSharedContext>::change_callbacks(self)
634    }
635
636    fn incoming_runtime(&self) -> &crate::subscriptions::incoming::IncomingRuntime {
637        <T as XmtpSharedContext>::incoming_runtime(self)
638    }
639
640    fn identity_resolution_registry(&self) -> &crate::identity_updates::IdentityResolutionRegistry {
641        <T as XmtpSharedContext>::identity_resolution_registry(self)
642    }
643
644    fn sync_metrics(&self) -> Option<Arc<WorkerMetrics<SyncMetric>>> {
645        <T as XmtpSharedContext>::sync_metrics(self)
646    }
647
648    fn mutexes(&self) -> &MutexRegistry {
649        <T as XmtpSharedContext>::mutexes(self)
650    }
651
652    fn cancellation_token(&self) -> &CancellationToken {
653        <T as XmtpSharedContext>::cancellation_token(self)
654    }
655
656    fn shutdown_complete(&self) -> bool {
657        <T as XmtpSharedContext>::shutdown_complete(self)
658    }
659
660    fn mark_shutdown_complete(&self) {
661        <T as XmtpSharedContext>::mark_shutdown_complete(self)
662    }
663
664    fn foreground_calls(&self) -> &Arc<ForegroundCalls> {
665        <T as XmtpSharedContext>::foreground_calls(self)
666    }
667
668    fn message_timestamp_ns(&self) -> i64 {
669        <T as XmtpSharedContext>::message_timestamp_ns(self)
670    }
671}
xmtp_mls/context.rs

xmtp_mls/
context.rs
