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

xmtp_api/
preflight.rs
