xmtp_mls_core/
preflight.rs

1//! Deferred admission and refresh share one client lifetime and commit lock.
2
3use crate::context::{CoreContext, LocalContext};
4use std::sync::{Arc, Weak, atomic::Ordering};
5use xmtp_api::preflight::{ConfigurationFetch, PreflightError, RequestPreflight};
6use xmtp_configuration::ServerConfiguration;
7use xmtp_mls_server_config::server_configuration::{
8    BlockedConnection, ConfigurationFetchError, ServerConfigError, check_minimum_version,
9    commit_configuration,
10};
11
12struct ClientPreflight<A, Db, S, F> {
13    context: Weak<LocalContext<A, Db, S, F>>,
14}
15
16/// Binds the client's request preflight to the API client once, keeping
17/// only a weak reference to the context.
18///
19/// # Panics
20///
21/// If a preflight is already bound.
22pub fn bind_preflight<A, Db, S, F>(context: &Arc<LocalContext<A, Db, S, F>>)
23where
24    A: xmtp_proto::api_client::XmtpApi + 'static,
25    Db: xmtp_db::XmtpDb + 'static,
26    S: xmtp_db::XmtpMlsStorageProvider + 'static,
27    F: xmtp_common::MaybeSend + xmtp_common::MaybeSync + 'static,
28{
29    let weak = Arc::downgrade(context);
30    assert!(
31        context.api().api_client.bind_preflight(|fetch| {
32            let _ = context.server_configuration().fetch.set(fetch);
33            Arc::new(ClientPreflight { context: weak })
34        }),
35        "client preflight is bound once"
36    );
37}
38
39#[xmtp_common::async_trait]
40impl<A, Db, S, F> RequestPreflight for ClientPreflight<A, Db, S, F>
41where
42    A: xmtp_proto::api_client::XmtpApi + 'static,
43    Db: xmtp_db::XmtpDb + 'static,
44    S: xmtp_db::XmtpMlsStorageProvider + 'static,
45    F: xmtp_common::MaybeSend + xmtp_common::MaybeSync + 'static,
46{
47    async fn check(&self, fetch: &dyn ConfigurationFetch) -> Result<(), PreflightError> {
48        let context = self
49            .context
50            .upgrade()
51            .ok_or_else(|| PreflightError::new(ServerConfigError::Closed))?;
52        run(&context, fetch, false)
53            .await
54            .map(|_| ())
55            .map_err(PreflightError::new)
56    }
57}
58
59fn check<C: CoreContext>(context: &C) -> Result<(), ServerConfigError> {
60    context.server_configuration().check()?;
61    if context.is_closed() {
62        return Err(ServerConfigError::Closed);
63    }
64    Ok(())
65}
66
67/// Fetches and commits the server configuration now.
68pub async fn refresh<C: CoreContext>(
69    context: &C,
70) -> Result<ServerConfiguration, ServerConfigError> {
71    let handle = context.server_configuration();
72    // Mock contexts do not bind a request adapter. Production builds always do.
73    let fetched = if let Some(fetch) = handle.fetch.get() {
74        run(context, fetch.as_ref(), true).await?
75    } else {
76        run(context, &UnboundFetch(context), true).await?
77    };
78    Ok(fetched.expect("refresh always fetches configuration"))
79}
80
81struct UnboundFetch<'a, C>(&'a C);
82#[xmtp_common::async_trait]
83impl<C: CoreContext> ConfigurationFetch for UnboundFetch<'_, C> {
84    async fn fetch(&self) -> xmtp_api::Result<xmtp_proto::backend_v1::GetConfigurationResponse> {
85        self.0.api().get_configuration().await
86    }
87    fn backend_url(&self) -> Option<&str> {
88        self.0.api().backend_url()
89    }
90}
91
92// implements: CONF-077
93async fn run<C: CoreContext>(
94    context: &C,
95    fetch: &dyn ConfigurationFetch,
96    refresh: bool,
97) -> Result<Option<ServerConfiguration>, ServerConfigError> {
98    let handle = context.server_configuration();
99    handle.check()?;
100    // Ready calls retain the caller's existing lifetime admission. Only a
101    // pending check or explicit refresh starts a cancellable foreground call.
102    if !refresh && !handle.pending.load(Ordering::Acquire) {
103        return Ok(None);
104    }
105    check(context)?;
106    let _foreground = context
107        .foreground_calls()
108        .enter()
109        .ok_or(ServerConfigError::Closed)?;
110    let cancelled = context.cancellation_token();
111    let _admission = tokio::select! {
112        biased;
113        _ = cancelled.cancelled() => { check(context)?; return Err(ServerConfigError::Closed); },
114        pending = handle.admission.lock() => pending,
115    };
116    check(context)?;
117    if !handle.pending.load(Ordering::Acquire) && !refresh {
118        return Ok(None);
119    }
120    let response = tokio::select! {
121        biased;
122        _ = cancelled.cancelled() => { check(context)?; return Err(ServerConfigError::Closed); },
123        response = fetch.fetch() => response.map_err(ConfigurationFetchError::Api)?,
124    };
125    check(context)?;
126    // No await from the binding read through the ready transition. The
127    // connection exists only for this commit, never for the network wait.
128    let fetched = commit_configuration(response, fetch.backend_url(), &context.db(), handle);
129    let fetched = match fetched {
130        Ok(fetched) => fetched,
131        Err(error) => {
132            if handle.blocked_connection().is_some() {
133                cancelled.cancel();
134            }
135            return Err(error);
136        }
137    };
138    // Store, then record the answer for its backend URL.
139    handle.record_deployment(&fetched.identifier).await?;
140    if let Err(ServerConfigError::Blocked(reason @ BlockedConnection::ClientVersionTooOld { .. })) =
141        check_minimum_version(&fetched, context.version_info().pkg_semver().semver())
142    {
143        let error = handle.block_connection(reason);
144        cancelled.cancel();
145        return Err(error.into());
146    }
147    handle.pending.store(false, Ordering::Release);
148    Ok(Some(fetched))
149}
xmtp_mls_core/preflight.rs

xmtp_mls_core/
preflight.rs
