xmtp_api/
preflight.rs

1//! Per-client admission before any backend request.
2
3use crate::{ApiError, Result, dyn_err};
4use std::sync::{
5    Arc, OnceLock,
6    atomic::{AtomicBool, Ordering},
7};
8use xmtp_common::{MaybeSend, MaybeSync, RetryableError};
9use xmtp_proto::{api::NetworkError, api_client::XmtpBackendClient, backend_v1 as wire};
10
11mod dispatch;
12
13/// A request failed before its target was sent. A later operation may retry.
14#[derive(Clone, Debug, thiserror::Error)]
15#[error("request configuration check failed: {cause}")]
16pub struct PreflightError {
17    #[source]
18    cause: Arc<NetworkError>,
19}
20impl PreflightError {
21    pub fn new(cause: impl RetryableError + 'static) -> Self {
22        Self {
23            cause: Arc::new(NetworkError::new(cause)),
24        }
25    }
26}
27impl RetryableError for PreflightError {
28    fn is_retryable(&self) -> bool {
29        self.cause.is_retryable()
30    }
31
32    fn rate_limit_delay(&self) -> Option<xmtp_common::time::Duration> {
33        self.cause.rate_limit_delay()
34    }
35}
36
37/// Find the admission marker without treating a failed configuration RPC as
38/// a failed target RPC. Explicitly unwrap boxes that hide their inner source.
39pub fn failure<'a>(error: &'a (dyn std::error::Error + 'static)) -> Option<&'a PreflightError> {
40    if let Some(error) = error.downcast_ref::<PreflightError>() {
41        return Some(error);
42    }
43    if let Some(error) = error.downcast_ref::<Box<ApiError>>() {
44        return failure(error.as_ref());
45    }
46    if let Some(error) = error.downcast_ref::<Box<PreflightError>>() {
47        return Some(error.as_ref());
48    }
49    error.source().and_then(failure)
50}
51
52/// Restricted access to the credential-free configuration RPC. The hook
53/// cannot use this capability to send the operation it is checking.
54#[xmtp_common::async_trait]
55pub trait ConfigurationFetch: MaybeSend + MaybeSync {
56    async fn fetch(&self) -> Result<wire::GetConfigurationResponse>;
57    fn backend_url(&self) -> Option<&str>;
58}
59
60#[xmtp_common::async_trait]
61pub trait RequestPreflight: MaybeSend + MaybeSync {
62    async fn check(
63        &self,
64        fetch: &dyn ConfigurationFetch,
65    ) -> std::result::Result<(), PreflightError>;
66}
67
68#[derive(Debug, thiserror::Error)]
69#[error("client configuration gate is not bound")]
70struct Unbound;
71impl RetryableError for Unbound {
72    fn is_retryable(&self) -> bool {
73        false
74    }
75}
76
77#[derive(Default)]
78struct Slot {
79    hook: OnceLock<Arc<dyn RequestPreflight>>,
80    pending: AtomicBool,
81}
82
83/// Backend dispatch with a clone-shared client admission hook. The raw
84/// transport is private so a caller cannot skip admission through this value.
85pub struct GuardedApi<A> {
86    raw: Arc<A>,
87    slot: Arc<Slot>,
88}
89impl<A> GuardedApi<A> {
90    pub(crate) fn new(raw: A) -> Self {
91        Self {
92            raw: Arc::new(raw),
93            slot: Arc::default(),
94        }
95    }
96    /// Mark the construction interval fail-closed until the client hook binds.
97    pub fn require_preflight(&self) {
98        self.slot.pending.store(true, Ordering::Release);
99    }
100    #[cfg(any(test, feature = "test-utils"))]
101    pub fn raw_for_test(&self) -> &A {
102        &self.raw
103    }
104    #[cfg(any(test, feature = "test-utils"))]
105    pub fn raw_mut_for_test(&mut self) -> Option<&mut A> {
106        Arc::get_mut(&mut self.raw)
107    }
108}
109impl<A: std::fmt::Debug> std::fmt::Debug for GuardedApi<A> {
110    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        f.debug_struct("GuardedApi")
112            .field("raw", &self.raw)
113            .finish_non_exhaustive()
114    }
115}
116impl<A> Clone for GuardedApi<A> {
117    fn clone(&self) -> Self {
118        Self {
119            raw: self.raw.clone(),
120            slot: self.slot.clone(),
121        }
122    }
123}
124struct RawConfiguration<A>(Arc<A>);
125#[xmtp_common::async_trait]
126impl<A: XmtpBackendClient> ConfigurationFetch for RawConfiguration<A> {
127    async fn fetch(&self) -> Result<wire::GetConfigurationResponse> {
128        self.0
129            .get_configuration(wire::GetConfigurationRequest {})
130            .await
131            .map_err(dyn_err)
132    }
133    fn backend_url(&self) -> Option<&str> {
134        self.0.backend_url()
135    }
136}
137impl<A: XmtpBackendClient> GuardedApi<A> {
138    /// Install the client hook with its restricted configuration capability.
139    pub fn bind_preflight(
140        &self,
141        make: impl FnOnce(Arc<dyn ConfigurationFetch>) -> Arc<dyn RequestPreflight>,
142    ) -> bool
143    where
144        A: 'static,
145    {
146        self.slot
147            .hook
148            .set(make(Arc::new(RawConfiguration(self.raw.clone()))))
149            .is_ok()
150    }
151    pub async fn check_preflight(&self) -> Result<()> {
152        if let Some(hook) = self.slot.hook.get() {
153            hook.check(&RawConfiguration(self.raw.clone()))
154                .await
155                .map_err(ApiError::Preflight)
156        } else if self.slot.pending.load(Ordering::Acquire) {
157            Err(ApiError::Preflight(PreflightError::new(Unbound)))
158        } else {
159            Ok(())
160        }
161    }
162}
163
164#[cfg(test)]
165mod tests;
xmtp_api/preflight.rs

xmtp_api/
preflight.rs
