xmtp_mls/server_configuration/
worker.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::client::ClientError;
16use crate::context::XmtpSharedContext;
17use crate::worker::{
18    BoxedWorker, NeedsDbReconnect, Worker, WorkerFactory, WorkerKind, WorkerResult,
19};
20
21use super::refresh;
22
23#[derive(Clone)]
24pub struct Factory<Context> {
25    context: Context,
26}
27
28impl<Context> WorkerFactory for Factory<Context>
29where
30    Context: XmtpSharedContext + 'static,
31{
32    fn kind(&self) -> WorkerKind {
33        WorkerKind::ConfigurationRefresh
34    }
35
36    fn create(
37        &self,
38        metrics: Option<crate::worker::DynMetrics>,
39    ) -> (BoxedWorker, Option<crate::worker::DynMetrics>) {
40        (
41            Box::new(ConfigurationWorker::new(self.context.clone())) as Box<_>,
42            metrics,
43        )
44    }
45}
46
47/// The refresh worker logs failures and leaves the stored copy alone.
48/// This exists only to satisfy the worker
49/// trait's error contract.
50#[derive(Debug, thiserror::Error)]
51#[error("the configuration refresh worker stopped")]
52pub struct ConfigurationWorkerError;
53
54impl NeedsDbReconnect for ConfigurationWorkerError {
55    fn needs_db_reconnect(&self) -> bool {
56        false
57    }
58}
59
60pub struct ConfigurationWorker<Context> {
61    context: Context,
62}
63
64impl<Context> ConfigurationWorker<Context> {
65    pub fn new(context: Context) -> Self {
66        Self { context }
67    }
68}
69
70#[xmtp_common::async_trait]
71impl<Context> Worker for ConfigurationWorker<Context>
72where
73    Context: XmtpSharedContext + 'static,
74{
75    fn kind(&self) -> WorkerKind {
76        WorkerKind::ConfigurationRefresh
77    }
78
79    async fn run_tasks(&mut self) -> WorkerResult<()> {
80        self.run().await;
81        Ok(())
82    }
83
84    fn factory<C>(context: C) -> impl WorkerFactory + 'static
85    where
86        C: XmtpSharedContext + 'static,
87    {
88        Factory { context }
89    }
90}
91
92impl<Context> ConfigurationWorker<Context>
93where
94    Context: XmtpSharedContext + 'static,
95{
96    async fn run(&mut self) {
97        loop {
98            let (base, jitter) = self.schedule();
99            xmtp_common::time::sleep(base + xmtp_common::time::rand_offset(jitter)).await;
100            // A client with a blocked connection has nothing left to learn from the backend.
101            if self
102                .context
103                .server_configuration()
104                .blocked_connection()
105                .is_some()
106            {
107                return;
108            }
109            self.tick().await;
110            // A blocked connection closes every open network stream. Cancelling
111            // is what closes them; the streams read the reason to report why.
112            if self
113                .context
114                .server_configuration()
115                .blocked_connection()
116                .is_some()
117            {
118                self.context.cancellation_token().cancel();
119                return;
120            }
121        }
122    }
123
124    /// `(base, jitter)`. A per-worker override wins; otherwise the compiled
125    /// hourly cadence and its 360-second spread.
126    fn schedule(&self) -> (Duration, Duration) {
127        let kind = WorkerKind::ConfigurationRefresh;
128        let (base, jitter) = self
129            .context
130            .worker_interval(kind, CONFIGURATION_REFRESH_INTERVAL);
131        let jitter = if self
132            .context
133            .worker_config()
134            .jitter_overrides
135            .contains_key(&kind)
136        {
137            jitter
138        } else {
139            CONFIGURATION_REFRESH_JITTER
140        };
141        (base, jitter)
142    }
143
144    /// One run: up to three attempts, then give up until the next run.
145    #[tracing::instrument(
146        skip_all,
147        fields(worker = "ConfigurationRefresh", operation = "worker_turn")
148    )]
149    pub(crate) async fn tick(&mut self) {
150        for attempt in 1..=CONFIGURATION_REFRESH_ATTEMPTS {
151            match self.attempt().await {
152                Ok(()) => return,
153                Err(error) => {
154                    // A server error never triggers another refresh.
155                    // The run's own schedule is the only thing driving this.
156                    tracing::warn!(
157                        attempt,
158                        %error,
159                        "server configuration refresh attempt failed; keeping the stored copy"
160                    );
161                    if !error.is_retryable() {
162                        return;
163                    }
164                    match CONFIGURATION_REFRESH_BACKOFF.get(attempt - 1) {
165                        Some(wait) => xmtp_common::time::sleep(*wait).await,
166                        None => return,
167                    }
168                }
169            }
170        }
171    }
172
173    // implements: CONF-036
174    async fn attempt(&mut self) -> Result<(), ClientError> {
175        refresh(&self.context).await.map(|_| ())
176    }
177}
xmtp_mls/server_configuration/worker.rs

xmtp_mls/server_configuration/
worker.rs
