xmtp_mls/
server_configuration.rs

1//! Resolving, storing, and refreshing what the backend publishes about itself.
2//!
3//! The client holds one immutable snapshot for its life
4//! and reads configuration through the provider, never the database. This
5//! module owns the three places the database is touched: the resolve that runs
6//! once inside `build`, the refresh worker, and the explicit refresh the SDKs
7//! expose.
8//!
9//! The snapshot carries the deployment's application component catalogue with
10//! everything else it publishes, so a refreshed catalogue, like any other
11//! refreshed value, reaches only a client created later on the database.
12
13use std::sync::{
14    Arc, OnceLock,
15    atomic::{AtomicBool, Ordering},
16};
17
18use parking_lot::RwLock;
19use prost::Message;
20use xmtp_api::{ApiClientWrapper, ApiError};
21use xmtp_configuration::{
22    BACKEND_DEFAULT_MAX_UPLOAD_BYTES, ConfigProvider, ServerConfiguration,
23    ServerConfigurationError, StoredConfigProvider,
24    attachments::{check_base_url, check_max_upload_bytes, check_retention_seconds},
25};
26use xmtp_db::prelude::*;
27use xmtp_db::{StorageError, server_configuration::StoredServerConfiguration};
28use xmtp_events::{ClientEvent, ClientRejectedByServer, EventWriter, RejectionCause};
29use xmtp_proto::api_client::XmtpBackendClient;
30use xmtp_proto::backend_v1;
31
32use crate::client::ClientError;
33
34/// Why a configuration read could not produce a usable copy.
35///
36/// Carried by [`ClientError::ConfigurationUnavailable`] so a caller can tell a
37/// backend that refused from a database that would not accept the answer.
38#[derive(Debug, thiserror::Error)]
39pub enum ConfigurationFetchError {
40    #[error("the deployment record could not be written: {0}")]
41    StorageLocation(#[from] crate::storage_location::StorageLocationError),
42    /// The backend did not answer, or answered with an error. A backend with
43    /// no `ConfigurationService` answers `UNIMPLEMENTED`; there is no shim.
44    #[error("the backend did not serve its configuration: {0}")]
45    Api(#[from] ApiError),
46    /// The answer arrived but could not be stored.
47    #[error("the fetched configuration could not be stored: {0}")]
48    Storage(#[from] StorageError),
49}
50
51impl xmtp_common::RetryableError for ConfigurationFetchError {
52    fn is_retryable(&self) -> bool {
53        match self {
54            Self::Api(e) => e.is_retryable(),
55            Self::Storage(e) => e.is_retryable(),
56            Self::StorageLocation(_) => false,
57        }
58    }
59}
60
61/// A condition that, once observed, fails later network work for the life of
62/// the client.
63#[derive(Clone, Debug, PartialEq, Eq)]
64pub enum BlockedConnection {
65    /// This database is bound to one deployment and another one answered.
66    BackendMismatch { stored: String, received: String },
67    /// The deployment now requires a newer client than this build.
68    ClientVersionTooOld { client: String, minimum: String },
69}
70
71impl From<&BlockedConnection> for ClientError {
72    fn from(blocked_connection: &BlockedConnection) -> Self {
73        match blocked_connection {
74            BlockedConnection::BackendMismatch { stored, received } => {
75                ClientError::BackendMismatch {
76                    stored: stored.clone(),
77                    received: received.clone(),
78                }
79            }
80            BlockedConnection::ClientVersionTooOld { client, minimum } => {
81                ClientError::ClientVersionTooOld {
82                    client: client.clone(),
83                    minimum: minimum.clone(),
84                }
85            }
86        }
87    }
88}
89
90/// The snapshot the client was built with, plus a blocked connection a refresh may set.
91///
92/// Cloning shares both: every consumer sees the blocked connection when a
93/// refresh sets it.
94#[derive(Clone)]
95pub struct ServerConfigurationHandle {
96    admission: Arc<tokio::sync::Mutex<()>>,
97    pending: Arc<AtomicBool>,
98    fetch: Arc<OnceLock<Arc<dyn xmtp_api::preflight::ConfigurationFetch>>>,
99    deployment_recorder: Arc<RwLock<Option<crate::storage_location::DeploymentRecorder>>>,
100    provider: Arc<dyn ConfigProvider>,
101    blocked_connection: Arc<RwLock<Option<BlockedConnection>>>,
102    event_writer: Arc<RwLock<Option<Arc<dyn EventWriter<()>>>>>,
103    /// The chains an app-supplied smart contract wallet signature may name.
104    /// `None` when the app supplied its own verifier.
105    restricted_chains: Option<Arc<[String]>>,
106}
107
108impl std::fmt::Debug for ServerConfigurationHandle {
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        f.debug_struct("ServerConfigurationHandle")
111            .field("identifier", &self.configuration().identifier)
112            .field("blocked_connection", &*self.blocked_connection.read())
113            .finish()
114    }
115}
116
117impl Default for ServerConfigurationHandle {
118    fn default() -> Self {
119        Self::new(Arc::new(StoredConfigProvider::default()))
120    }
121}
122
123impl ServerConfigurationHandle {
124    #[cfg(any(test, feature = "test-utils"))]
125    pub(crate) fn custom_verifier(&self) -> bool {
126        self.restricted_chains.is_none()
127    }
128
129    pub(crate) fn requires_preflight(&self) -> bool {
130        self.pending.load(Ordering::Acquire)
131    }
132
133    /// Hold one snapshot, with no zero left in its limits or attachment upload ceiling.
134    ///
135    /// Wire conversion replaces a zero on the wire with the compiled default, but a
136    /// snapshot an app builds in Rust and hands in through a `ConfigProvider`
137    /// never passes through that conversion, and a zero dimension
138    /// would panic the `chunks(limit)` calls in `xmtp_api`. Every
139    /// snapshot reaches a client through this constructor, so sanitizing here
140    /// keeps zero limits out of all three readers at once: this handle, the
141    /// wrapper that chunks with them, and the transport. It also removes an
142    /// unusable attachment offer from a provider snapshot.
143    pub fn new(provider: Arc<dyn ConfigProvider>) -> Self {
144        let sanitized = {
145            let supplied = provider.server_configuration();
146            let limits = supplied.limits.without_zeroes();
147            let attachments = supplied.attachments.as_ref().and_then(|offer| {
148                let mut offer = offer.clone();
149                if offer.max_upload_bytes == 0 {
150                    offer.max_upload_bytes = BACKEND_DEFAULT_MAX_UPLOAD_BYTES;
151                }
152                let checked = check_base_url(&offer.base_url)
153                    .and_then(|_| check_max_upload_bytes(offer.max_upload_bytes))
154                    .and_then(|_| check_retention_seconds(offer.retention_seconds));
155                if let Err(reason) = checked {
156                    tracing::warn!(
157                        field = reason.field(),
158                        reason = reason.reason(),
159                        "ignoring unusable attachment storage offer"
160                    );
161                    None
162                } else {
163                    Some(offer)
164                }
165            });
166            (limits != supplied.limits || attachments != supplied.attachments).then(|| {
167                ServerConfiguration {
168                    limits,
169                    attachments,
170                    ..supplied.clone()
171                }
172            })
173        };
174        Self {
175            deployment_recorder: Arc::default(),
176            provider: match sanitized {
177                Some(configuration) => Arc::new(StoredConfigProvider::new(configuration)),
178                None => provider,
179            },
180            blocked_connection: Arc::default(),
181            event_writer: Arc::default(),
182            restricted_chains: None,
183            admission: Arc::default(),
184            pending: Arc::default(),
185            fetch: Arc::default(),
186        }
187    }
188
189    /// Replace the test catalogue and keep resolved admission and preflight state.
190    #[cfg(any(test, feature = "test-utils"))]
191    pub(crate) fn with_application_components(
192        mut self,
193        components: Vec<xmtp_configuration::ApplicationComponentDefinition>,
194    ) -> Self {
195        let mut configuration = self.configuration().clone();
196        configuration.application_components = components;
197        self.provider = Arc::new(StoredConfigProvider::new(configuration));
198        self
199    }
200
201    /// Restrict app-supplied smart contract wallet signatures to the chains the
202    /// snapshot names. Skipped entirely when the app
203    /// supplied its own verifier, which the chain restriction exempts.
204    pub(crate) fn with_chain_restriction(mut self, custom_verifier: bool) -> Self {
205        self.restricted_chains = (!custom_verifier).then(|| {
206            Arc::<[String]>::from(self.configuration().smart_contract_wallet_chains.clone())
207        });
208        self
209    }
210
211    /// Bind a signature request to the chains this deployment accepts, before
212    /// it is handed to the app.
213    pub fn restrict(&self, request: &mut xmtp_id::associations::builder::SignatureRequest) {
214        if let Some(chains) = self.restricted_chains.clone() {
215            request.restrict_chains(chains);
216        }
217    }
218
219    /// The snapshot. Fixed for the life of the client: a refresh
220    /// rewrites the stored row, never this value.
221    pub fn configuration(&self) -> &ServerConfiguration {
222        self.provider.server_configuration()
223    }
224
225    /// Whether this deployment keeps a commit log. Read at every
226    /// write and read site, so a deployment that turns it off turns it off for
227    /// every group this client touches.
228    pub fn commit_log_enabled(&self) -> bool {
229        self.configuration().mls.commit_log_enabled()
230    }
231
232    /// The blocked connection, if one has been observed.
233    pub fn blocked_connection(&self) -> Option<BlockedConnection> {
234        self.blocked_connection.read().clone()
235    }
236
237    pub(crate) fn set_event_writer(&self, writer: Arc<dyn EventWriter<()>>) {
238        *self.event_writer.write() = Some(writer);
239    }
240
241    pub(crate) fn set_deployment_recorder(
242        &self,
243        recorder: crate::storage_location::DeploymentRecorder,
244    ) {
245        *self.deployment_recorder.write() = Some(recorder);
246    }
247
248    async fn record_deployment(&self, identifier: &str) -> Result<(), ClientError> {
249        let recorder = self.deployment_recorder.read().clone();
250        if let Some(recorder) = recorder {
251            recorder.record(identifier).await.map_err(|error| {
252                ClientError::ConfigurationUnavailable(Box::new(
253                    ConfigurationFetchError::StorageLocation(error),
254                ))
255            })?;
256        }
257        Ok(())
258    }
259
260    /// Fail when the connection is blocked. Every call that reaches the network goes
261    /// through here.
262    // implements: CONF-075
263    pub fn check(&self) -> Result<(), ClientError> {
264        match self.blocked_connection.read().as_ref() {
265            Some(blocked_connection) => Err(blocked_connection.into()),
266            None => Ok(()),
267        }
268    }
269
270    /// Replace a failure with the blocked connection's error when a block is
271    /// recorded. A block during an operation, such as the deferred first
272    /// request check, cancels the client's streams, so the stopped work
273    /// would otherwise report a close or an incomplete barrier instead of why.
274    // implements: CONF-075, CONF-077
275    pub(crate) fn blocked_or<E: From<ClientError>>(&self, error: E) -> E {
276        match self.check() {
277            Err(blocked) => blocked.into(),
278            Ok(()) => error,
279        }
280    }
281
282    /// Block the connection on the first failure. A later one does not displace it: the
283    /// first cause is the one worth reporting.
284    pub(crate) fn block_connection(&self, reason: BlockedConnection) -> ClientError {
285        let mut guard = self.blocked_connection.write();
286        if let Some(held) = guard.as_ref() {
287            return held.into();
288        }
289        let error = ClientError::from(&reason);
290        let event = match &reason {
291            BlockedConnection::BackendMismatch { .. } => ClientRejectedByServer {
292                cause: RejectionCause::BackendMismatch,
293                min_libxmtp_version: None,
294            },
295            BlockedConnection::ClientVersionTooOld { minimum, .. } => ClientRejectedByServer {
296                cause: RejectionCause::VersionTooOld,
297                min_libxmtp_version: Some(minimum.clone()),
298            },
299        };
300        *guard = Some(reason);
301        drop(guard);
302        if let Some(writer) = self.event_writer.read().as_ref() {
303            writer.emit(Some(ClientEvent::ClientRejectedByServer(event)), None);
304        }
305        error
306    }
307}
308
309/// Decode a stored row into a snapshot.
310///
311/// A copy that does not decode is a warning, not a failure. The
312/// identifier stays available for the binding check and every value falls back
313/// to its compiled default.
314fn snapshot_from(stored: &StoredServerConfiguration) -> ServerConfiguration {
315    match backend_v1::GetConfigurationResponse::decode(stored.response.as_slice()) {
316        Ok(response) => ServerConfiguration::from(response),
317        Err(error) => {
318            tracing::warn!(
319                identifier = %stored.identifier,
320                %error,
321                "stored server configuration does not decode; using compiled defaults"
322            );
323            ServerConfiguration {
324                identifier: stored.identifier.clone(),
325                ..Default::default()
326            }
327        }
328    }
329}
330
331/// Validate a fetched response and turn it into a snapshot.
332pub(crate) fn validated(
333    response: &backend_v1::GetConfigurationResponse,
334) -> Result<ServerConfiguration, ServerConfigurationError> {
335    let configuration = ServerConfiguration::from(response.clone());
336    configuration.validate()?;
337    Ok(configuration)
338}
339
340/// Fetch, validate, and store one copy, applying the identifier binding.
341///
342/// Shared by the build-time fetch, the refresh worker,
343/// and the explicit refresh the SDKs expose.
344// implements: CONF-030, CONF-040, CONF-071
345pub(crate) async fn fetch_and_store<ApiClient>(
346    api: &ApiClientWrapper<ApiClient>,
347    db: &impl DbQuery,
348    handle: &ServerConfigurationHandle,
349) -> Result<ServerConfiguration, ClientError>
350where
351    ApiClient: XmtpBackendClient,
352{
353    handle.check()?;
354
355    let response = api.get_configuration().await.map_err(|error| {
356        ClientError::ConfigurationUnavailable(Box::new(ConfigurationFetchError::Api(error)))
357    })?;
358    let configuration = commit_configuration(response, api.backend_url(), db, handle)?;
359    handle.record_deployment(&configuration.identifier).await?;
360    Ok(configuration)
361}
362
363fn commit_configuration(
364    response: backend_v1::GetConfigurationResponse,
365    backend_url: Option<&str>,
366    db: &impl DbQuery,
367    handle: &ServerConfigurationHandle,
368) -> Result<ServerConfiguration, ClientError> {
369    let configuration = validated(&response)?;
370
371    // The identifier, not the URL, is the binding. A row with an
372    // empty identifier was written by an offline build and binds nothing.
373    let stored = db.server_configuration().map_err(storage_unavailable)?;
374    if let Some(stored) = stored.as_ref()
375        && !stored.identifier.is_empty()
376        && stored.identifier != configuration.identifier
377    {
378        tracing::error!(
379            stored = %stored.identifier,
380            received = %configuration.identifier,
381            "this database is bound to a different backend deployment"
382        );
383        // A failure to record it keeps the connection blocked in memory, so this
384        // client still fails every later call.
385        if let Err(error) = db.record_server_configuration_conflict(&configuration.identifier) {
386            tracing::error!(%error, "could not record the conflicting backend identifier");
387        }
388        return Err(handle.block_connection(BlockedConnection::BackendMismatch {
389            stored: stored.identifier.clone(),
390            received: configuration.identifier.clone(),
391        }));
392    }
393
394    db.store_server_configuration(
395        &configuration.identifier,
396        normalized_url(backend_url.unwrap_or_default()),
397        &response.encode_to_vec(),
398        xmtp_common::time::now_ns(),
399    )
400    .map_err(storage_unavailable)?;
401
402    Ok(configuration)
403}
404
405/// Read a deployment's configuration with no database and no client.
406///
407/// The one call that needs no credential, so an app can learn whether
408/// the deployment requires authentication, which scopes it wants, and which
409/// chains it accepts before it decides how to build a client. Nothing is
410/// stored and no identifier binding is applied: there is no database to bind to.
411pub async fn fetch_server_configuration<ApiClient>(
412    api: &ApiClientWrapper<ApiClient>,
413) -> Result<ServerConfiguration, ClientError>
414where
415    ApiClient: XmtpBackendClient,
416{
417    let response = api.get_configuration().await.map_err(|error| {
418        ClientError::ConfigurationUnavailable(Box::new(ConfigurationFetchError::Api(error)))
419    })?;
420    Ok(validated(&response)?)
421}
422
423/// The form a backend URL is stored and compared in.
424///
425/// A transport reports the URI it dialled, which for `http://host:port` carries
426/// a trailing slash the app never typed. Normalising both sides keeps a purely
427/// cosmetic difference from looking like a move to another deployment.
428pub(crate) fn normalized_url(url: &str) -> &str {
429    url.trim_end_matches('/')
430}
431
432fn storage_unavailable(error: StorageError) -> ClientError {
433    ClientError::ConfigurationUnavailable(Box::new(ConfigurationFetchError::Storage(error)))
434}
435
436/// Resolve the snapshot `build` will hold.
437///
438/// The minimum-version check is not here: it applies to
439/// the snapshot the client ends up holding, which may have come from a
440/// caller-supplied provider that never reached this function.
441///
442/// Runs before any identity work. Offline, it never awaits a network call, so
443/// `build_offline` still completes without polling a pending future.
444// implements: CONF-026, CONF-027, CONF-033, CONF-034, CONF-072
445pub(crate) async fn resolve<ApiClient>(
446    api: &ApiClientWrapper<ApiClient>,
447    db: &impl DbQuery,
448    allow_offline: bool,
449) -> Result<ServerConfigurationHandle, ClientError>
450where
451    ApiClient: XmtpBackendClient,
452{
453    let stored = db.server_configuration().map_err(storage_unavailable)?;
454
455    // A recorded conflict fails every build until the database is
456    // replaced with one created for the backend the app now uses.
457    if let Some(conflicting) = stored
458        .as_ref()
459        .and_then(|row| row.conflicting_identifier.clone())
460    {
461        return Err(ClientError::BackendMismatch {
462            stored: stored.map(|row| row.identifier).unwrap_or_default(),
463            received: conflicting,
464        });
465    }
466
467    let pending = allow_offline
468        && stored.as_ref().is_some_and(|stored| {
469            api.backend_url()
470                .is_some_and(|url| normalized_url(url) != normalized_url(&stored.backend_url))
471        });
472    let configuration = match (allow_offline, stored) {
473        // Offline with no copy is compiled defaults and an empty
474        // identifier. The first successful refresh writes the row.
475        (true, None) => ServerConfiguration::default(),
476        // Offline with a copy uses it, and skips the URL check.
477        (true, Some(stored)) => snapshot_from(&stored),
478        // Online with no copy fetches before any identity work.
479        (false, None) => {
480            let handle = ServerConfigurationHandle::default();
481            fetch_and_store(api, db, &handle).await?
482        }
483        (false, Some(stored)) => {
484            // The stored copy is used as is unless the
485            // configured URL moved, in which case the identifier is checked
486            // again before anything else happens.
487            let moved = api
488                .backend_url()
489                .is_some_and(|url| normalized_url(url) != normalized_url(&stored.backend_url));
490            if moved {
491                tracing::info!(
492                    from = %stored.backend_url,
493                    to = api.backend_url().unwrap_or_default(),
494                    "backend URL changed; re-reading the deployment configuration"
495                );
496                let handle = ServerConfigurationHandle::default();
497                fetch_and_store(api, db, &handle).await?
498            } else {
499                snapshot_from(&stored)
500            }
501        }
502    };
503
504    let mut handle =
505        ServerConfigurationHandle::new(Arc::new(StoredConfigProvider::new(configuration)));
506    handle.pending = Arc::new(AtomicBool::new(pending));
507    Ok(handle)
508}
509
510/// Compare on major, minor, and patch only, so a
511/// prerelease tag never makes a client too old.
512// implements: CONF-049
513pub(crate) fn check_minimum_version(
514    configuration: &ServerConfiguration,
515    client_version: &semver::Version,
516) -> Result<(), ClientError> {
517    let Some(minimum) = configuration.minimum_version()? else {
518        return Ok(());
519    };
520    if xmtp_configuration::version_is_below(client_version, &minimum) {
521        return Err(ClientError::ClientVersionTooOld {
522            client: client_version.to_string(),
523            minimum: minimum.to_string(),
524        });
525    }
526    Ok(())
527}
528
529pub mod worker;
530
531#[cfg(test)]
532mod tests;
533
534mod preflight;
535pub(crate) use preflight::{bind_preflight, refresh};
xmtp_mls/server_configuration.rs

xmtp_mls/
server_configuration.rs
