xmtp_mls_core/
context.rs

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

xmtp_mls_core/
context.rs
