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

xmtp_mls_server_config/
server_configuration.rs
