xmtp_mls_core/
config_refresh.rs

1//! The hourly refresh.
2//!
3//! A run rewrites the stored row. It never changes the snapshot the client is
4//! holding — that is fixed at build — so the only thing a refresh can
5//! change about a running client is to block its connection on a failure it must not ignore: a
6//! different deployment answering or a minimum version this build no
7//! longer meets.
8
9use xmtp_common::{RetryableError, time::Duration};
10use xmtp_configuration::{
11    CONFIGURATION_REFRESH_ATTEMPTS, CONFIGURATION_REFRESH_BACKOFF, CONFIGURATION_REFRESH_INTERVAL,
12    CONFIGURATION_REFRESH_JITTER,
13};
14
15use crate::context::CoreContext;
16use crate::worker::{
17    BoxedWorker, NeedsDbReconnect, NewWorker, Worker, WorkerFactory, WorkerKind, WorkerResult,
18};
19use xmtp_mls_server_config::server_configuration::ServerConfigError;
20
21use super::preflight::refresh;
22
23#[derive(Clone)]
24pub struct Factory<Context> {
25    context: Context,
26}
27
28impl<Context> WorkerFactory for Factory<Context>
29where
30    Context: CoreContext + 'static,
31{
32    fn kind(&self) -> WorkerKind {
33        WorkerKind::ConfigurationRefresh
34    }
35
36    fn create(&self) -> BoxedWorker {
37        Box::new(ConfigurationWorker::new(self.context.clone()))
38    }
39}
40
41/// The refresh worker logs failures and leaves the stored copy alone.
42/// This exists only to satisfy the worker
43/// trait's error contract.
44#[derive(Debug, thiserror::Error)]
45#[error("the configuration refresh worker stopped")]
46pub struct ConfigurationWorkerError;
47
48impl NeedsDbReconnect for ConfigurationWorkerError {
49    fn needs_db_reconnect(&self) -> bool {
50        false
51    }
52}
53
54pub struct ConfigurationWorker<Context> {
55    context: Context,
56}
57
58impl<Context> ConfigurationWorker<Context> {
59    pub fn new(context: Context) -> Self {
60        Self { context }
61    }
62}
63
64#[xmtp_common::async_trait]
65impl<Context> Worker for ConfigurationWorker<Context>
66where
67    Context: CoreContext + 'static,
68{
69    fn kind(&self) -> WorkerKind {
70        WorkerKind::ConfigurationRefresh
71    }
72
73    async fn run_tasks(&mut self) -> WorkerResult<()> {
74        self.run().await;
75        Ok(())
76    }
77}
78
79impl<Context> NewWorker<Context> for ConfigurationWorker<Context>
80where
81    Context: CoreContext + 'static,
82{
83    fn factory(context: Context) -> impl WorkerFactory + 'static {
84        Factory { context }
85    }
86}
87
88impl<Context> ConfigurationWorker<Context>
89where
90    Context: CoreContext + 'static,
91{
92    async fn run(&mut self) {
93        loop {
94            let (base, jitter) = self.schedule();
95            xmtp_common::time::sleep(base + xmtp_common::time::rand_offset(jitter)).await;
96            // A client with a blocked connection has nothing left to learn from the backend.
97            if self
98                .context
99                .server_configuration()
100                .blocked_connection()
101                .is_some()
102            {
103                return;
104            }
105            self.tick().await;
106            // A blocked connection closes every open network stream. Cancelling
107            // is what closes them; the streams read the reason to report why.
108            if self
109                .context
110                .server_configuration()
111                .blocked_connection()
112                .is_some()
113            {
114                self.context.cancellation_token().cancel();
115                return;
116            }
117        }
118    }
119
120    /// `(base, jitter)`. A per-worker override wins; otherwise the compiled
121    /// hourly cadence and its 360-second spread.
122    fn schedule(&self) -> (Duration, Duration) {
123        let kind = WorkerKind::ConfigurationRefresh;
124        let (base, jitter) = self
125            .context
126            .worker_interval(kind, CONFIGURATION_REFRESH_INTERVAL);
127        let jitter = if self
128            .context
129            .worker_config()
130            .jitter_overrides
131            .contains_key(&kind)
132        {
133            jitter
134        } else {
135            CONFIGURATION_REFRESH_JITTER
136        };
137        (base, jitter)
138    }
139
140    /// One run: up to three attempts, then give up until the next run.
141    #[tracing::instrument(
142        skip_all,
143        fields(worker = "ConfigurationRefresh", operation = "worker_turn")
144    )]
145    pub(crate) async fn tick(&mut self) {
146        for attempt in 1..=CONFIGURATION_REFRESH_ATTEMPTS {
147            match self.attempt().await {
148                Ok(()) => return,
149                Err(error) => {
150                    // A server error never triggers another refresh.
151                    // The run's own schedule is the only thing driving this.
152                    tracing::warn!(
153                        attempt,
154                        %error,
155                        "server configuration refresh attempt failed; keeping the stored copy"
156                    );
157                    if !error.is_retryable() {
158                        return;
159                    }
160                    match CONFIGURATION_REFRESH_BACKOFF.get(attempt - 1) {
161                        Some(wait) => xmtp_common::time::sleep(*wait).await,
162                        None => return,
163                    }
164                }
165            }
166        }
167    }
168
169    // implements: CONF-036
170    async fn attempt(&mut self) -> Result<(), ServerConfigError> {
171        refresh(&self.context).await.map(|_| ())
172    }
173}
174
175#[cfg(any(test, feature = "test-utils"))]
176#[doc(hidden)]
177pub mod test_support {
178    use super::ConfigurationWorker;
179    use crate::context::CoreContext;
180
181    /// Run one refresh of `worker`, as its schedule does.
182    pub async fn tick<Context>(worker: &mut ConfigurationWorker<Context>)
183    where
184        Context: CoreContext + 'static,
185    {
186        worker.tick().await
187    }
188}
189
190#[cfg(test)]
191#[cfg(not(target_arch = "wasm32"))]
192mod tests;
xmtp_mls_core/config_refresh.rs

xmtp_mls_core/
config_refresh.rs
