xmtp_mls/server_configuration/
worker.rs

1//! The hourly refresh (CFG-046 to CFG-050).
2//!
3//! A run rewrites the stored row. It never changes the snapshot the client is
4//! holding — that is fixed at build (CFG-030) — so the only thing a refresh can
5//! change about a running client is to latch a failure it must not ignore: a
6//! different deployment answering (CFG-051) or a minimum version this build no
7//! longer meets (CFG-061).
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::{ConfigurationLatch, check_minimum_version, fetch_and_store};
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 never surfaces an error: CFG-047 says a failed run logs
48/// and leaves the stored copy alone. 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 latched client has nothing left to learn from the backend.
101            if self.context.server_configuration().latched().is_some() {
102                return;
103            }
104            self.tick().await;
105            // CFG-051 and CFG-061: a latch closes every open stream. Cancelling
106            // is what closes them; the streams read the latch to report why.
107            if self.context.server_configuration().latched().is_some() {
108                self.context.cancellation_token().cancel();
109                return;
110            }
111        }
112    }
113
114    /// `(base, jitter)`. A per-worker override wins; otherwise the compiled
115    /// hourly cadence and its 360-second spread.
116    fn schedule(&self) -> (Duration, Duration) {
117        let kind = WorkerKind::ConfigurationRefresh;
118        let (base, jitter) = self
119            .context
120            .worker_interval(kind, CONFIGURATION_REFRESH_INTERVAL);
121        let jitter = if self
122            .context
123            .worker_config()
124            .jitter_overrides
125            .contains_key(&kind)
126        {
127            jitter
128        } else {
129            CONFIGURATION_REFRESH_JITTER
130        };
131        (base, jitter)
132    }
133
134    /// One run: up to three attempts, then give up until the next run.
135    #[tracing::instrument(
136        skip_all,
137        fields(worker = "ConfigurationRefresh", operation = "worker_turn")
138    )]
139    pub(crate) async fn tick(&mut self) {
140        for attempt in 1..=CONFIGURATION_REFRESH_ATTEMPTS {
141            match self.attempt().await {
142                Ok(()) => return,
143                Err(error) => {
144                    // CFG-049: a server error never triggers another refresh.
145                    // The run's own schedule is the only thing driving this.
146                    tracing::warn!(
147                        attempt,
148                        %error,
149                        "server configuration refresh attempt failed; keeping the stored copy"
150                    );
151                    if !error.is_retryable() {
152                        return;
153                    }
154                    match CONFIGURATION_REFRESH_BACKOFF.get(attempt - 1) {
155                        Some(wait) => xmtp_common::time::sleep(*wait).await,
156                        None => return,
157                    }
158                }
159            }
160        }
161    }
162
163    async fn attempt(&mut self) -> Result<(), ClientError> {
164        let handle = self.context.server_configuration();
165        let db = self.context.db();
166        let fetched = fetch_and_store(self.context.api(), &db, handle).await?;
167
168        // CFG-061: the copy is stored either way, and the client stops.
169        if let Err(ClientError::ClientVersionTooOld { client, minimum }) =
170            check_minimum_version(&fetched, self.context.version_info().pkg_semver().semver())
171        {
172            tracing::error!(
173                %client,
174                %minimum,
175                "the backend now requires a newer libxmtp than this client"
176            );
177            handle.latch(ConfigurationLatch::ClientVersionTooOld { client, minimum });
178        }
179        Ok(())
180    }
181}
xmtp_mls/server_configuration/worker.rs

xmtp_mls/server_configuration/
worker.rs
