agentmux_srv/muxbus/wan_presence/
machine.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The presence publisher's decisions, apart from its I/O. Given what just
5//! happened ([`Event`]) and the time ([`Now`]), [`Machine::next`] settles the
6//! state, says what to do ([`Action`]) and how long to wait before the next
7//! [`Event::Due`]. Nothing here reads a clock, a random number or the
8//! network, so every rule is tested with a fake clock; `super::Driver` runs
9//! it against the relay.
10//!
11//! | state | entered on | next try |
12//! |---|---|---|
13//! | `signed_out` | no MuxBus sign-in | a sign-in (a local recheck every minute, no request) |
14//! | `publishing` | the record was stored | a minute, or a change of the snapshot |
15//! | `retrying` | unreachable, 5xx, 429, any other status | 5 s doubling to 5 min, full jitter; a 429's `Retry-After` |
16//! | `unsupported` | 404, no presence route | about 5 min; at once on a sign-in, a network change, a wake, or a new relay version |
17//! | `rejected` | 400 or 410 | 10 min or more |
18//! | `signed_off` | a goodbye was stored at a sign-out | a sign-in |
19//! | `off` | this install doesn't publish (`super::policy`) | turned on |
20//!
21//! Outside `publishing` a change of the snapshot only marks it dirty: the
22//! next try sends the latest one, no sooner. A 409 means the relay already
23//! has a newer record of this install's, which is as good as storing this
24//! one.
25//!
26//! A goodbye (a v3 record, `gone`) is due when something was published and
27//! the relay takes v2 or later ([`Machine::farewell`]): an older relay would
28//! refuse v3, so it gets none. Turning publishing off says it here
29//! ([`Action::Goodbye`]); a sign-out and a quit say it from outside, and a
30//! sign-out tells the machine with [`Event::Goodbye`].
31//!
32//! Two refusals get an immediate second try before `rejected`: a record the
33//! relay can't parse is sent again one version down, v4 then v2 then v1 (an
34//! older relay knows no agent lines, an older one still no states), and the
35//! version it took is then kept for that relay for [`Config::v1_memory`]; a
36//! record whose time the relay refuses is sent again stamped with the
37//! relay's clock, learned from the `Date` of every answer.
38
39use std::collections::HashMap;
40use std::time::Duration;
41
42use agentmux_common::install_presence::{PRESENCE_VERSION, PRESENCE_VERSION_LINES, PRESENCE_VERSION_V1};
43
44/// The version a fresh try sends to a relay not known to be older.
45pub(crate) const LIVE_VERSION: u32 = PRESENCE_VERSION_LINES;
46
47/// The version to try after `v` was refused as malformed, if any.
48fn step_down(v: u32) -> Option<u32> {
49    match v {
50        PRESENCE_VERSION_LINES => Some(PRESENCE_VERSION),
51        PRESENCE_VERSION => Some(PRESENCE_VERSION_V1),
52        _ => None,
53    }
54}
55
56use crate::backend::rpc_types::{PresenceOffReason, PresenceState, PresenceStatusResult};
57
58/// The relay's `error` for a record it can't parse, which is what a relay
59/// that predates the record's version answers.
60pub(crate) const MALFORMED_RECORD: &str = "Malformed install presence record";
61
62/// A clock further than this from the relay's is shown in the status.
63const CLOCK_NOTE_AFTER_MS: u64 = 2 * 60 * 1000;
64
65/// The waits, apart so tests can shorten them.
66#[derive(Debug, Clone)]
67pub(crate) struct Config {
68    /// Between two publishes of an unchanged record.
69    pub interval: Duration,
70    /// The first retry's ceiling; it doubles per failure.
71    pub backoff_base: Duration,
72    pub backoff_cap: Duration,
73    /// The least wait between retries, so full jitter never spins.
74    pub backoff_floor: Duration,
75    /// The longest `Retry-After` honoured.
76    pub retry_after_cap: Duration,
77    /// After a 404, jittered by a fifth either way.
78    pub unsupported_retry: Duration,
79    /// How often the relay's version is checked while `unsupported`.
80    pub health_every: Duration,
81    /// After a refusal, plus up to a fifth.
82    pub rejected_retry: Duration,
83    /// How often a signed-out publisher looks for a sign-in, locally.
84    pub signed_out_recheck: Duration,
85    /// How long a relay that took only v1 gets v1.
86    pub v1_memory: Duration,
87    /// An outage this long is logged as a warning, once.
88    pub outage_warn_after: Duration,
89}
90
91pub(crate) const LIVE: Config = Config {
92    interval: Duration::from_secs(60),
93    backoff_base: Duration::from_secs(5),
94    backoff_cap: Duration::from_secs(5 * 60),
95    backoff_floor: Duration::from_secs(1),
96    retry_after_cap: Duration::from_secs(60 * 60),
97    unsupported_retry: Duration::from_secs(5 * 60),
98    health_every: Duration::from_secs(60),
99    rejected_retry: Duration::from_secs(10 * 60),
100    signed_out_recheck: Duration::from_secs(60),
101    v1_memory: Duration::from_secs(6 * 60 * 60),
102    outage_warn_after: Duration::from_secs(10 * 60),
103};
104
105/// The time, as the driver read it. `mono_ms` only ever grows and drives the
106/// waits; `wall_ms` is this computer's clock, for the status and the record;
107/// `rand` is uniform in `[0, 1)`, for jitter.
108#[derive(Debug, Clone, Copy)]
109pub(crate) struct Now {
110    pub mono_ms: u64,
111    pub wall_ms: u64,
112    pub rand: f64,
113}
114
115/// What the relay made of one record.
116#[derive(Debug, Clone, PartialEq, Eq)]
117pub(crate) enum Answer {
118    /// 2xx.
119    Stored,
120    /// 404: the relay has no presence route.
121    NoRoute,
122    /// 400 [`MALFORMED_RECORD`].
123    Malformed,
124    /// 400 for the record's time: the relay's message.
125    ClockRefused(String),
126    /// Any other 400, a 410, or a record that couldn't be built: why.
127    Refused(String),
128    /// 429.
129    TooMany { retry_after: Option<Duration> },
130    /// Unreachable, 5xx or any other status: a short reason and the detail.
131    Unavailable { reason: String, detail: String },
132}
133
134#[derive(Debug, Clone, PartialEq, Eq)]
135pub(crate) enum Event {
136    /// The publisher started.
137    Start,
138    /// The last wait ran out.
139    Due,
140    /// The snapshot changed (while publishing: after the debounce).
141    Changed,
142    /// A try found no sign-in; nothing was sent.
143    NoSignIn,
144    /// A record of version `sent_v` was sent. `relay_date_ms` is the answer's
145    /// `Date`, when it had a readable one.
146    Answer { sent_v: u32, answer: Answer, relay_date_ms: Option<u64> },
147    /// The relay's health check: its version, `None` when it failed.
148    Health { version: Option<String>, relay_date_ms: Option<u64> },
149    /// A sign-in, a sign-out or a new account.
150    SignIn,
151    /// This computer's network addresses changed.
152    NetworkChanged,
153    /// This computer woke from sleep.
154    Woke,
155    /// "Publish now": forget the backoff and try at once.
156    PublishNow,
157    /// Whether this install publishes: `None` to publish, else why not.
158    Policy { off: Option<PresenceOffReason> },
159    /// The settings changed: the driver reads the policy again and passes
160    /// it on as [`Event::Policy`]; the machine itself ignores this.
161    SettingsChanged,
162    /// A goodbye went out (from [`Action::Goodbye`], or a sign-out); `stored`
163    /// when the relay took it.
164    Goodbye { stored: bool },
165}
166
167#[derive(Debug, Clone, Copy, PartialEq, Eq)]
168pub(crate) enum Action {
169    /// Nothing; wait [`Step::wait`].
170    Wait,
171    /// Send the latest snapshot as a record of version `v`.
172    Attempt { v: u32 },
173    /// Ask the relay its version.
174    CheckHealth,
175    /// Send the goodbye, stamped `published_at_ms`, then report
176    /// [`Event::Goodbye`].
177    Goodbye { published_at_ms: u64 },
178}
179
180/// A goodbye is due, stamped on the relay's clock as far as it is known.
181#[derive(Debug, Clone, Copy, PartialEq, Eq)]
182pub(crate) struct Farewell {
183    offset_ms: i64,
184}
185
186impl Farewell {
187    /// `published_at_ms` for a goodbye sent at `wall_ms`.
188    pub(crate) fn stamp(&self, wall_ms: u64) -> u64 {
189        u64::try_from(wall_ms as i64 + self.offset_ms).unwrap_or(0)
190    }
191}
192
193#[derive(Debug, Clone, PartialEq, Eq)]
194pub(crate) struct Log {
195    pub warn: bool,
196    pub text: String,
197}
198
199#[derive(Debug, Clone, PartialEq, Eq)]
200pub(crate) struct Step {
201    pub action: Action,
202    /// Until the next [`Event::Due`]; meaningful when `action` is `Wait`.
203    pub wait: Duration,
204    pub logs: Vec<Log>,
205}
206
207/// The second tries already spent on the record being sent.
208#[derive(Debug, Default, Clone, Copy)]
209struct Followups {
210    /// The record was sent again one version down at least once.
211    stepped_down: bool,
212    clock: bool,
213}
214
215#[derive(Debug)]
216pub(crate) struct Machine {
217    config: Config,
218    relay: String,
219    /// `None` until the first state is entered.
220    state: Option<PresenceState>,
221    since_wall_ms: u64,
222    reason: Option<String>,
223    last_ok_wall_ms: Option<u64>,
224    failures: u32,
225    due_ms: u64,
226    health_due_ms: Option<u64>,
227    dirty: bool,
228    followups: Followups,
229    /// Relays that took only an older version: which, and until when.
230    older_until_ms: HashMap<String, (u32, u64)>,
231    offset_ms: Option<i64>,
232    relay_version: Option<String>,
233    /// The relay's version just after its 404.
234    version_at_404: Option<String>,
235    outage_since_ms: Option<u64>,
236    outage_warned: bool,
237    /// Why this install doesn't publish; `None` while it does.
238    off: Option<PresenceOffReason>,
239    /// This machine asked for a goodbye ([`Action::Goodbye`]) whose result
240    /// hasn't come back yet. A sign-out reports [`Event::Goodbye`] too, even
241    /// when none was due, so only this says the result is worth logging.
242    goodbye_sent: bool,
243}
244
245fn ms(d: Duration) -> u64 {
246    u64::try_from(d.as_millis()).unwrap_or(u64::MAX)
247}
248
249fn scaled(d: Duration, factor: f64) -> u64 {
250    (ms(d) as f64 * factor) as u64
251}
252
253/// "12 min", rounded, at least 1.
254fn minutes(ms: u64) -> u64 {
255    ((ms + 30_000) / 60_000).max(1)
256}
257
258impl Machine {
259    pub(crate) fn new(config: Config, relay: &str) -> Self {
260        Self {
261            config,
262            relay: relay.trim_end_matches('/').to_string(),
263            state: None,
264            since_wall_ms: 0,
265            reason: None,
266            last_ok_wall_ms: None,
267            failures: 0,
268            due_ms: 0,
269            health_due_ms: None,
270            dirty: false,
271            followups: Followups::default(),
272            older_until_ms: HashMap::new(),
273            offset_ms: None,
274            relay_version: None,
275            version_at_404: None,
276            outage_since_ms: None,
277            outage_warned: false,
278            off: None,
279            goodbye_sent: false,
280        }
281    }
282
283    #[cfg(test)]
284    pub(crate) fn state(&self) -> Option<PresenceState> {
285        self.state
286    }
287
288    /// Whether a snapshot change is published (after the debounce), rather
289    /// than only marked dirty.
290    pub(crate) fn follows_changes(&self) -> bool {
291        self.state == Some(PresenceState::Publishing)
292    }
293
294    #[cfg(test)]
295    pub(crate) fn dirty(&self) -> bool {
296        self.dirty
297    }
298
299    #[cfg(test)]
300    pub(crate) fn set_relay(&mut self, relay: &str) {
301        self.relay = relay.trim_end_matches('/').to_string();
302    }
303
304    /// `published_at_ms` for a record sent at `wall_ms`: on the relay's
305    /// clock when an answer has told it, else this computer's.
306    pub(crate) fn stamp(&self, wall_ms: u64) -> u64 {
307        Farewell { offset_ms: self.offset_ms.unwrap_or(0) }.stamp(wall_ms)
308    }
309
310    /// Whether a goodbye is due if this install went away now: something
311    /// was published, publishing is on and no goodbye has been said since,
312    /// and the relay takes v2 or later (one that took only v1 would refuse v3).
313    pub(crate) fn farewell(&self, now: &Now) -> Option<Farewell> {
314        let due = self.off.is_none()
315            && self.state != Some(PresenceState::SignedOff)
316            && self.last_ok_wall_ms.is_some()
317            && self.record_version(now) >= PRESENCE_VERSION;
318        due.then(|| Farewell { offset_ms: self.offset_ms.unwrap_or(0) })
319    }
320
321    /// The version a fresh try sends: an older one while this relay is
322    /// remembered as taking only that.
323    pub(crate) fn record_version(&self, now: &Now) -> u32 {
324        match self.older_until_ms.get(&self.relay) {
325            Some(&(v, until)) if now.mono_ms < until => v,
326            _ => LIVE_VERSION,
327        }
328    }
329
330    pub(crate) fn next(&mut self, event: Event, now: Now) -> Step {
331        let mut logs = Vec::new();
332        let action = if self.off.is_some() {
333            self.next_while_off(event, &now, &mut logs)
334        } else {
335            self.next_while_on(event, now, &mut logs)
336        };
337        self.warn_long_outage(&now, &mut logs);
338        let due = match self.health_due_ms {
339            Some(h) if self.state == Some(PresenceState::Unsupported) => self.due_ms.min(h),
340            _ => self.due_ms,
341        };
342        Step { action, wait: Duration::from_millis(due.saturating_sub(now.mono_ms)), logs }
343    }
344
345    /// Off: nothing is sent until publishing is turned on again; a goodbye
346    /// in flight when it was turned off lands here.
347    fn next_while_off(&mut self, event: Event, now: &Now, logs: &mut Vec<Log>) -> Action {
348        match event {
349            Event::Policy { off: None } => {
350                self.off = None;
351                self.failures = 0;
352                logs.push(Log { warn: false, text: "publishing is on".into() });
353                self.attempt(None, now)
354            }
355            Event::Policy { off: Some(reason) } => {
356                self.enter_off(reason, now, logs);
357                Action::Wait
358            }
359            Event::Goodbye { stored } => {
360                let reason = self.off.unwrap_or(PresenceOffReason::Setting);
361                if std::mem::take(&mut self.goodbye_sent) {
362                    logs.push(Log { warn: false, text: goodbye_text(stored).into() });
363                }
364                self.enter_off(reason, now, logs);
365                Action::Wait
366            }
367            Event::Changed => {
368                self.dirty = true;
369                Action::Wait
370            }
371            _ => Action::Wait,
372        }
373    }
374
375    /// Stop publishing for `reason`; one log line per new reason.
376    fn enter_off(&mut self, reason: PresenceOffReason, now: &Now, logs: &mut Vec<Log>) {
377        let changed = self.off != Some(reason) || self.state != Some(PresenceState::Off);
378        self.off = Some(reason);
379        self.failures = 0;
380        self.dirty = false;
381        let text = format!("off: {}", off_text(reason));
382        let before = logs.len();
383        self.enter(PresenceState::Off, None, &text, now, logs);
384        if changed && logs.len() == before {
385            logs.push(Log { warn: false, text });
386        }
387        self.due_ms = now.mono_ms + ms(self.config.interval);
388    }
389
390    fn next_while_on(&mut self, event: Event, now: Now, logs: &mut Vec<Log>) -> Action {
391        match event {
392            Event::Policy { off: None } | Event::SettingsChanged => {
393                if self.state.is_none() {
394                    self.attempt(None, &now)
395                } else {
396                    Action::Wait
397                }
398            }
399            Event::Policy { off: Some(reason) } => match self.farewell(&now) {
400                Some(farewell) => {
401                    self.off = Some(reason);
402                    self.goodbye_sent = true;
403                    Action::Goodbye { published_at_ms: farewell.stamp(now.wall_ms) }
404                }
405                None => {
406                    self.enter_off(reason, &now, logs);
407                    Action::Wait
408                }
409            },
410            Event::Goodbye { stored: true } => {
411                self.failures = 0;
412                self.enter(PresenceState::SignedOff, None, goodbye_text(true), &now, logs);
413                self.due_ms = now.mono_ms + ms(self.config.signed_out_recheck);
414                Action::Wait
415            }
416            Event::Goodbye { stored: false } => {
417                self.failures = 0;
418                self.attempt(None, &now)
419            }
420            Event::Start => self.attempt(None, &now),
421            Event::PublishNow | Event::SignIn => {
422                self.failures = 0;
423                self.attempt(None, &now)
424            }
425            Event::Due => {
426                let health_first = self.state == Some(PresenceState::Unsupported) && now.mono_ms < self.due_ms;
427                if health_first {
428                    self.health_due_ms = None;
429                    Action::CheckHealth
430                } else {
431                    self.attempt(None, &now)
432                }
433            }
434            Event::Changed => {
435                if self.follows_changes() {
436                    self.attempt(None, &now)
437                } else {
438                    self.dirty = true;
439                    Action::Wait
440                }
441            }
442            Event::NetworkChanged | Event::Woke => match self.state {
443                Some(PresenceState::Publishing | PresenceState::Retrying | PresenceState::Unsupported) => {
444                    self.attempt(None, &now)
445                }
446                _ => Action::Wait,
447            },
448            Event::NoSignIn => {
449                self.failures = 0;
450                // After a goodbye, "signed off" says more than "signed out".
451                if self.state != Some(PresenceState::SignedOff) {
452                    let text = "not signed in to MuxBus, nothing is published";
453                    self.enter(PresenceState::SignedOut, None, text, &now, logs);
454                }
455                self.due_ms = now.mono_ms + ms(self.config.signed_out_recheck);
456                Action::Wait
457            }
458            Event::Health { version, relay_date_ms } => {
459                self.observe_date(relay_date_ms, &now, logs);
460                self.on_health(version, &now, logs)
461            }
462            Event::Answer { sent_v, answer, relay_date_ms } => {
463                self.observe_date(relay_date_ms, &now, logs);
464                self.on_answer(sent_v, answer, &now, logs)
465            }
466        }
467    }
468
469    /// Send the latest snapshot. `v` is set for a second try of the same
470    /// record; a fresh try forgets the second tries spent.
471    fn attempt(&mut self, v: Option<u32>, now: &Now) -> Action {
472        if v.is_none() {
473            self.followups = Followups::default();
474        }
475        self.dirty = false;
476        Action::Attempt { v: v.unwrap_or_else(|| self.record_version(now)) }
477    }
478
479    fn on_answer(&mut self, sent_v: u32, answer: Answer, now: &Now, logs: &mut Vec<Log>) -> Action {
480        match answer {
481            Answer::Stored => {
482                if self.followups.stepped_down {
483                    let until = now.mono_ms + ms(self.config.v1_memory);
484                    self.older_until_ms.insert(self.relay.clone(), (sent_v, until));
485                }
486                self.failures = 0;
487                self.last_ok_wall_ms = Some(now.wall_ms);
488                let text = match sent_v {
489                    PRESENCE_VERSION_V1 => "published, as v1: the relay is on an older version and gets no agent states",
490                    PRESENCE_VERSION => "published, as v2: the relay is on an older version and gets no agent summaries",
491                    _ => "published",
492                };
493                self.enter(PresenceState::Publishing, None, text, now, logs);
494                self.due_ms = now.mono_ms + ms(self.config.interval);
495                Action::Wait
496            }
497            Answer::Malformed if step_down(sent_v).is_some() => {
498                self.followups.stepped_down = true;
499                self.attempt(step_down(sent_v), now)
500            }
501            Answer::ClockRefused(_) if !self.followups.clock => {
502                self.followups.clock = true;
503                self.attempt(Some(sent_v), now)
504            }
505            Answer::Malformed => self.reject(MALFORMED_RECORD.to_string(), now, logs),
506            Answer::ClockRefused(why) | Answer::Refused(why) => self.reject(why, now, logs),
507            Answer::NoRoute => {
508                self.enter(
509                    PresenceState::Unsupported,
510                    Some("the relay has no presence route yet".into()),
511                    "the relay has no presence route yet, checking again within minutes",
512                    now,
513                    logs,
514                );
515                self.due_ms = now.mono_ms + scaled(self.config.unsupported_retry, 0.8 + 0.4 * now.rand);
516                // A version to compare later ones with, read now.
517                self.version_at_404 = None;
518                self.health_due_ms = None;
519                Action::CheckHealth
520            }
521            Answer::TooMany { retry_after } => {
522                self.failures = self.failures.saturating_add(1);
523                let wait = match retry_after {
524                    Some(d) => ms(d.min(self.config.retry_after_cap)).max(ms(self.config.backoff_floor)),
525                    None => self.backoff(now),
526                };
527                let reason = "too many requests".to_string();
528                let text = format!("retrying: {reason}");
529                self.enter(PresenceState::Retrying, Some(reason), &text, now, logs);
530                self.due_ms = now.mono_ms + wait;
531                Action::Wait
532            }
533            Answer::Unavailable { reason, detail } => {
534                self.failures = self.failures.saturating_add(1);
535                let wait = self.backoff(now);
536                let text = format!("retrying: {reason} ({detail})");
537                self.enter(PresenceState::Retrying, Some(reason), &text, now, logs);
538                self.due_ms = now.mono_ms + wait;
539                Action::Wait
540            }
541        }
542    }
543
544    fn reject(&mut self, why: String, now: &Now, logs: &mut Vec<Log>) -> Action {
545        let text = format!("the relay refused the record: {why}");
546        self.enter(PresenceState::Rejected, Some(why), &text, now, logs);
547        self.due_ms = now.mono_ms + scaled(self.config.rejected_retry, 1.0 + 0.2 * now.rand);
548        Action::Wait
549    }
550
551    fn on_health(&mut self, version: Option<String>, now: &Now, logs: &mut Vec<Log>) -> Action {
552        if let Some(v) = &version {
553            self.relay_version = Some(v.clone());
554        }
555        if self.state != Some(PresenceState::Unsupported) {
556            return Action::Wait;
557        }
558        match (version, &self.version_at_404) {
559            (Some(v), Some(before)) if v != *before => {
560                logs.push(Log { warn: false, text: format!("the relay moved from {before} to {v}, trying again") });
561                self.attempt(None, now)
562            }
563            (Some(v), None) => {
564                self.version_at_404 = Some(v);
565                self.health_due_ms = Some(now.mono_ms + ms(self.config.health_every));
566                Action::Wait
567            }
568            _ => {
569                self.health_due_ms = Some(now.mono_ms + ms(self.config.health_every));
570                Action::Wait
571            }
572        }
573    }
574
575    /// Full jitter: anywhere up to `base * 2^(failures - 1)`, capped, but no
576    /// less than the floor.
577    fn backoff(&self, now: &Now) -> u64 {
578        let doublings = self.failures.saturating_sub(1).min(20);
579        let ceiling = ms(self.config.backoff_base).saturating_mul(1 << doublings).min(ms(self.config.backoff_cap));
580        ((ceiling as f64 * now.rand) as u64).max(ms(self.config.backoff_floor))
581    }
582
583    /// Learn the relay's clock from an answer's `Date`.
584    fn observe_date(&mut self, relay_date_ms: Option<u64>, now: &Now, logs: &mut Vec<Log>) {
585        let Some(date) = relay_date_ms else { return };
586        let skewed_before = self.clock_skewed();
587        self.offset_ms = Some(date as i64 - now.wall_ms as i64);
588        if self.clock_skewed() && !skewed_before {
589            let off = self.offset_ms.unwrap_or(0).unsigned_abs();
590            logs.push(Log {
591                warn: false,
592                text: format!("this computer's clock differs from the relay's by {} min; records carry the relay's time", minutes(off)),
593            });
594        }
595    }
596
597    fn clock_skewed(&self) -> bool {
598        self.offset_ms.is_some_and(|o| o.unsigned_abs() > CLOCK_NOTE_AFTER_MS)
599    }
600
601    /// Enter `state` for `reason`; one log line when the state changes.
602    fn enter(&mut self, state: PresenceState, reason: Option<String>, text: &str, now: &Now, logs: &mut Vec<Log>) {
603        let failing = matches!(state, PresenceState::Retrying | PresenceState::Unsupported | PresenceState::Rejected);
604        if failing {
605            self.outage_since_ms.get_or_insert(now.mono_ms);
606        } else {
607            self.outage_since_ms = None;
608            self.outage_warned = false;
609        }
610        if state != PresenceState::Unsupported {
611            self.health_due_ms = None;
612        }
613        self.reason = reason;
614        if self.state != Some(state) {
615            self.state = Some(state);
616            self.since_wall_ms = now.wall_ms;
617            logs.push(Log { warn: false, text: text.to_string() });
618        }
619    }
620
621    fn warn_long_outage(&mut self, now: &Now, logs: &mut Vec<Log>) {
622        let Some(since) = self.outage_since_ms else { return };
623        let lasted = now.mono_ms.saturating_sub(since);
624        if self.outage_warned || lasted < ms(self.config.outage_warn_after) {
625            return;
626        }
627        self.outage_warned = true;
628        let why = self.reason.as_deref().unwrap_or("unknown");
629        logs.push(Log { warn: true, text: format!("not published for {} min: {why}", minutes(lasted)) });
630    }
631
632    /// The status for `presence.status`; `None` before the first state.
633    pub(crate) fn status(&self, now: &Now) -> Option<PresenceStatusResult> {
634        let state = self.state?;
635        let waits_for_nothing =
636            matches!(state, PresenceState::SignedOut | PresenceState::SignedOff | PresenceState::Off);
637        let next_try_ms = (!waits_for_nothing).then(|| now.wall_ms + self.due_ms.saturating_sub(now.mono_ms));
638        let mut notes = Vec::new();
639        if self.clock_skewed() {
640            let off = self.offset_ms.unwrap_or(0).unsigned_abs();
641            notes.push(format!("Your clock differs from the cloud's by {} min", minutes(off)));
642        }
643        let record_version = self.record_version(now);
644        if record_version == PRESENCE_VERSION_V1 {
645            notes.push("The cloud runs an older version, so your devices don't see agent states".to_string());
646        } else if record_version == PRESENCE_VERSION {
647            notes.push("The cloud runs an older version, so your devices don't see agent summaries".to_string());
648        }
649        Some(PresenceStatusResult {
650            state,
651            since_ms: self.since_wall_ms,
652            last_ok_ms: self.last_ok_wall_ms,
653            last_error: self.reason.clone(),
654            next_try_ms,
655            relay_version: self.relay_version.clone(),
656            offset_ms: self.offset_ms,
657            record_version,
658            note: (!notes.is_empty()).then(|| notes.join(". ")),
659            off_reason: if state == PresenceState::Off { self.off } else { None },
660        })
661    }
662}
663
664fn goodbye_text(stored: bool) -> &'static str {
665    if stored {
666        "said goodbye: devices show this computer as offline"
667    } else {
668        "the goodbye didn't reach the relay; devices notice in a few minutes"
669    }
670}
671
672fn off_text(reason: PresenceOffReason) -> &'static str {
673    match reason {
674        PresenceOffReason::Setting => "turned off in Settings",
675        PresenceOffReason::DevBuild => "a dev build doesn't publish",
676        PresenceOffReason::Headless => "a headless install doesn't publish",
677        PresenceOffReason::IsolatedHome => "an isolated home doesn't publish",
678        PresenceOffReason::TestHarness => "an install under test doesn't publish",
679    }
680}
681
682/// What the relay's status and `error` mean for the publisher.
683pub(crate) fn classify(status: u16, error: Option<&str>, retry_after: Option<Duration>) -> Answer {
684    let said = || error.unwrap_or_default().to_string();
685    match status {
686        200..=299 => Answer::Stored,
687        // Newest wins on the relay: it has a newer record of this install's.
688        409 => Answer::Stored,
689        404 => Answer::NoRoute,
690        400 if error == Some(MALFORMED_RECORD) => Answer::Malformed,
691        // The relay names the field when it refuses the record's time.
692        400 if error.is_some_and(|e| e.contains("published_at_ms")) => Answer::ClockRefused(said()),
693        400 | 410 => Answer::Refused(error.map_or_else(|| format!("the relay answered {status}"), str::to_string)),
694        429 => Answer::TooMany { retry_after },
695        _ => Answer::Unavailable { reason: format!("the cloud answered {status}"), detail: said() },
696    }
697}
698
699#[cfg(test)]
700mod tests {
701    use super::*;
702
703    const S: u64 = 1000;
704    const MIN: u64 = 60 * S;
705
706    /// A machine at mono 0, wall `WALL`, that has just stored a record.
707    const WALL: u64 = 1_791_000_000_000;
708
709    fn at(mono_ms: u64) -> Now {
710        Now { mono_ms, wall_ms: WALL + mono_ms, rand: 0.5 }
711    }
712
713    fn at_rand(mono_ms: u64, rand: f64) -> Now {
714        Now { rand, ..at(mono_ms) }
715    }
716
717    fn answer(answer: Answer) -> Event {
718        Event::Answer { sent_v: 4, answer, relay_date_ms: None }
719    }
720
721    fn unreachable() -> Event {
722        answer(Answer::Unavailable { reason: "cloud unreachable".into(), detail: "connection refused".into() })
723    }
724
725    fn machine() -> Machine {
726        Machine::new(LIVE, "https://relay.test")
727    }
728
729    fn publishing() -> Machine {
730        let mut m = machine();
731        assert_eq!(m.next(Event::Start, at(0)).action, Action::Attempt { v: 4 });
732        m.next(answer(Answer::Stored), at(0));
733        m
734    }
735
736    fn secs(d: Duration) -> u64 {
737        d.as_secs()
738    }
739
740    #[test]
741    fn a_stored_record_publishes_again_in_a_minute() {
742        let mut m = machine();
743        let step = m.next(answer(Answer::Stored), at(0));
744        assert_eq!(m.state(), Some(PresenceState::Publishing));
745        assert_eq!((step.action, step.wait), (Action::Wait, Duration::from_secs(60)));
746        assert_eq!(step.logs, vec![Log { warn: false, text: "published".into() }]);
747        assert_eq!(m.next(Event::Due, at(60 * S)).action, Action::Attempt { v: 4 });
748    }
749
750    #[test]
751    fn while_publishing_a_change_is_sent_after_the_drivers_debounce() {
752        let mut m = publishing();
753        assert!(m.follows_changes());
754        assert_eq!(m.next(Event::Changed, at(5 * S)).action, Action::Attempt { v: 4 });
755    }
756
757    #[test]
758    fn no_sign_in_sends_nothing_and_rechecks_locally() {
759        let mut m = machine();
760        let step = m.next(Event::NoSignIn, at(0));
761        assert_eq!(m.state(), Some(PresenceState::SignedOut));
762        assert_eq!((step.action, step.wait), (Action::Wait, Duration::from_secs(60)));
763        assert_eq!(m.status(&at(0)).unwrap().next_try_ms, None, "no try is promised while signed out");
764        assert_eq!(m.next(Event::Changed, at(S)).action, Action::Wait, "a change while signed out sends nothing");
765        assert_eq!(m.next(Event::Woke, at(2 * S)).action, Action::Wait);
766        assert_eq!(m.next(Event::SignIn, at(3 * S)).action, Action::Attempt { v: 4 }, "a sign-in tries at once");
767    }
768
769    #[test]
770    fn failures_back_off_exponentially_to_five_minutes() {
771        let mut m = machine();
772        // rand just under 1: the ceiling itself.
773        let ceilings: Vec<u64> = (0..9).map(|i| secs(m.next(unreachable(), at_rand(i * S, 0.999_999)).wait)).collect();
774        assert_eq!(ceilings, vec![4, 9, 19, 39, 79, 159, 299, 299, 299], "5 s, 10 s, 20 s ... capped at 300 s, less the rounding");
775        assert_eq!(m.state(), Some(PresenceState::Retrying));
776    }
777
778    #[test]
779    fn backoff_is_full_jitter_with_a_floor() {
780        for (rand, expect_ms) in [(0.0, 1000), (0.1, 1000), (0.5, 2500), (0.9, 4500)] {
781            let mut m = machine();
782            assert_eq!(m.next(unreachable(), at_rand(0, rand)).wait, Duration::from_millis(expect_ms), "rand {rand}");
783        }
784        // The fourth failure: every draw inside [floor, 40 s].
785        for i in 0..100 {
786            let mut probe = Machine { failures: 3, ..machine() };
787            let wait = probe.next(unreachable(), at_rand(0, i as f64 / 100.0)).wait;
788            assert!(wait >= Duration::from_secs(1) && wait <= Duration::from_secs(40), "{wait:?}");
789        }
790    }
791
792    #[test]
793    fn a_success_resets_the_backoff() {
794        let mut m = machine();
795        for i in 0..5 {
796            m.next(unreachable(), at(i));
797        }
798        m.next(answer(Answer::Stored), at(10 * S));
799        let wait = m.next(unreachable(), at_rand(20 * S, 0.999_999)).wait;
800        assert_eq!(secs(wait), 4, "back to the first step");
801    }
802
803    #[test]
804    fn a_429_honours_retry_after_and_otherwise_backs_off() {
805        let mut m = machine();
806        let step = m.next(answer(Answer::TooMany { retry_after: Some(Duration::from_secs(90)) }), at(0));
807        assert_eq!(m.state(), Some(PresenceState::Retrying));
808        assert_eq!(step.wait, Duration::from_secs(90), "Retry-After past the backoff ceiling is still honoured");
809        let step = m.next(answer(Answer::TooMany { retry_after: None }), at_rand(0, 0.999_999));
810        assert_eq!(secs(step.wait), 9, "the second failure's backoff");
811        let step = m.next(answer(Answer::TooMany { retry_after: Some(Duration::from_secs(86_400)) }), at(0));
812        assert_eq!(step.wait, Duration::from_secs(3600), "a day is capped");
813        assert_eq!(m.status(&at(0)).unwrap().last_error.as_deref(), Some("too many requests"));
814    }
815
816    #[test]
817    fn a_404_waits_about_five_minutes_not_an_hour() {
818        for rand in [0.0, 0.5, 0.999_999] {
819            let mut m = machine();
820            let step = m.next(answer(Answer::NoRoute), at_rand(0, rand));
821            assert_eq!(m.state(), Some(PresenceState::Unsupported));
822            assert_eq!(step.action, Action::CheckHealth, "the relay's version is read at once");
823            let step = m.next(Event::Health { version: Some("1.0.0".into()), relay_date_ms: None }, at_rand(0, rand));
824            assert_eq!(step.wait, Duration::from_secs(60), "the next version check comes first");
825            // Version checks every minute until the retry is due.
826            let mut t = 0;
827            loop {
828                t += 60 * S;
829                let step = m.next(Event::Due, at(t));
830                if step.action != Action::CheckHealth {
831                    assert_eq!(step.action, Action::Attempt { v: 4 });
832                    break;
833                }
834                m.next(Event::Health { version: Some("1.0.0".into()), relay_date_ms: None }, at(t));
835            }
836        }
837        // The retry itself lands in [4 min, 6 min].
838        for rand in [0.0, 0.25, 0.5, 0.75, 0.999_999] {
839            let mut m = machine();
840            m.next(answer(Answer::NoRoute), at_rand(0, rand));
841            let retry_ms = m.status(&at(0)).unwrap().next_try_ms.unwrap() - WALL;
842            assert!((4 * MIN..=6 * MIN).contains(&retry_ms), "rand {rand}: {retry_ms}");
843        }
844    }
845
846    #[test]
847    fn a_404_is_cut_short_by_a_sign_in_a_network_change_a_wake_or_publish_now() {
848        for event in [Event::SignIn, Event::NetworkChanged, Event::Woke, Event::PublishNow] {
849            let mut m = machine();
850            m.next(answer(Answer::NoRoute), at(0));
851            m.next(Event::Health { version: Some("1.0.0".into()), relay_date_ms: None }, at(0));
852            assert_eq!(m.next(event.clone(), at(5 * S)).action, Action::Attempt { v: 4 }, "{event:?}");
853        }
854    }
855
856    #[test]
857    fn a_404_is_cut_short_by_a_new_relay_version() {
858        let mut m = machine();
859        m.next(answer(Answer::NoRoute), at(0));
860        m.next(Event::Health { version: Some("1.0.0".into()), relay_date_ms: None }, at(0));
861        assert_eq!(m.next(Event::Due, at(60 * S)).action, Action::CheckHealth);
862        let step = m.next(Event::Health { version: Some("1.0.0".into()), relay_date_ms: None }, at(60 * S));
863        assert_eq!(step.action, Action::Wait, "the same version: keep waiting");
864        assert_eq!(m.next(Event::Due, at(120 * S)).action, Action::CheckHealth);
865        let step = m.next(Event::Health { version: None, relay_date_ms: None }, at(120 * S));
866        assert_eq!(step.action, Action::Wait, "a failed check changes nothing");
867        assert_eq!(m.next(Event::Due, at(180 * S)).action, Action::CheckHealth);
868        let step = m.next(Event::Health { version: Some("1.1.0".into()), relay_date_ms: None }, at(180 * S));
869        assert_eq!(step.action, Action::Attempt { v: 4 });
870        assert_eq!(m.status(&at(180 * S)).unwrap().relay_version.as_deref(), Some("1.1.0"));
871        // Still 404 on the new version: that version is the new baseline.
872        assert_eq!(m.next(answer(Answer::NoRoute), at(181 * S)).action, Action::CheckHealth);
873        let step = m.next(Event::Health { version: Some("1.1.0".into()), relay_date_ms: None }, at(181 * S));
874        assert_eq!(step.action, Action::Wait);
875    }
876
877    #[test]
878    fn version_checks_come_at_most_once_a_minute() {
879        let mut m = machine();
880        m.next(answer(Answer::NoRoute), at(0));
881        m.next(Event::Health { version: Some("1".into()), relay_date_ms: None }, at(0));
882        // A change in between brings the next check no closer.
883        assert_eq!(m.next(Event::Changed, at(10 * S)).wait, Duration::from_secs(50));
884    }
885
886    #[test]
887    fn a_change_while_failing_only_marks_the_record_dirty() {
888        for setup in [unreachable(), answer(Answer::NoRoute), answer(Answer::Refused("nope".into()))] {
889            let mut m = machine();
890            m.next(setup.clone(), at(0));
891            let before = m.next(Event::Changed, at(S));
892            assert_eq!(before.action, Action::Wait, "{setup:?}");
893            assert!(m.dirty());
894            let again = m.next(Event::Changed, at(2 * S));
895            assert_eq!(again.wait + Duration::from_secs(1), before.wait, "the wait runs on, not restarted");
896            assert!(!m.follows_changes());
897        }
898        let mut m = machine();
899        m.next(unreachable(), at(0));
900        m.next(Event::Changed, at(100));
901        let step = m.next(Event::Due, at(10 * S));
902        assert_eq!(step.action, Action::Attempt { v: 4 }, "the next try sends the latest");
903        assert!(!m.dirty());
904    }
905
906    #[test]
907    fn a_refusal_is_retried_at_most_every_ten_minutes() {
908        for rand in [0.0, 0.5, 0.999_999] {
909            let mut m = machine();
910            let step = m.next(answer(Answer::Refused("not signed by its instance".into())), at_rand(0, rand));
911            assert_eq!(m.state(), Some(PresenceState::Rejected));
912            assert!(step.wait >= Duration::from_secs(600) && step.wait <= Duration::from_secs(720), "{:?}", step.wait);
913            assert_eq!(m.next(Event::Woke, at(S)).action, Action::Wait, "a wake doesn't fix a refusal");
914            assert_eq!(m.next(Event::NetworkChanged, at(S)).action, Action::Wait);
915            assert_eq!(m.status(&at(S)).unwrap().last_error.as_deref(), Some("not signed by its instance"));
916        }
917        let mut m = machine();
918        m.next(answer(Answer::Refused("x".into())), at(0));
919        assert_eq!(m.next(Event::PublishNow, at(S)).action, Action::Attempt { v: 4 }, "but the user may ask");
920    }
921
922    #[test]
923    fn a_malformed_record_is_sent_again_one_version_down_until_v1() {
924        let mut m = publishing();
925        assert_eq!(m.next(Event::Due, at(60 * S)).action, Action::Attempt { v: 4 });
926        let step = m.next(answer(Answer::Malformed), at(60 * S));
927        assert_eq!(step.action, Action::Attempt { v: 2 });
928        assert_eq!(m.state(), Some(PresenceState::Publishing), "no state change for the second try");
929        let step = m.next(Event::Answer { sent_v: 2, answer: Answer::Malformed, relay_date_ms: None }, at(60 * S));
930        assert_eq!(step.action, Action::Attempt { v: 1 });
931        let step = m.next(Event::Answer { sent_v: 1, answer: Answer::Malformed, relay_date_ms: None }, at(60 * S));
932        assert_eq!(step.action, Action::Wait, "v1 failing too: no third try");
933        assert_eq!(m.state(), Some(PresenceState::Rejected));
934        assert_eq!(m.status(&at(60 * S)).unwrap().record_version, 4, "nothing remembered");
935    }
936
937    #[test]
938    fn a_relay_that_took_v1_gets_v1_for_six_hours() {
939        let mut m = machine();
940        m.next(Event::Start, at(0));
941        assert_eq!(m.next(answer(Answer::Malformed), at(0)).action, Action::Attempt { v: 2 });
942        let malformed_v2 = Event::Answer { sent_v: 2, answer: Answer::Malformed, relay_date_ms: None };
943        assert_eq!(m.next(malformed_v2, at(0)).action, Action::Attempt { v: 1 });
944        let step = m.next(Event::Answer { sent_v: 1, answer: Answer::Stored, relay_date_ms: None }, at(0));
945        assert_eq!(m.state(), Some(PresenceState::Publishing));
946        assert!(step.logs[0].text.contains("as v1"), "{:?}", step.logs);
947        let status = m.status(&at(0)).unwrap();
948        assert_eq!(status.record_version, 1);
949        assert!(status.note.as_deref().unwrap().contains("older version"), "{status:?}");
950
951        assert_eq!(m.next(Event::Due, at(60 * S)).action, Action::Attempt { v: 1 });
952        m.next(Event::Answer { sent_v: 1, answer: Answer::Stored, relay_date_ms: None }, at(60 * S));
953        // Per relay: another one gets v4.
954        m.set_relay("https://other.test/");
955        assert_eq!(m.next(Event::Due, at(120 * S)).action, Action::Attempt { v: 4 });
956        m.set_relay("https://relay.test");
957        assert_eq!(m.next(Event::Due, at(6 * 60 * MIN - 1)).action, Action::Attempt { v: 1 });
958        assert_eq!(m.next(Event::Due, at(6 * 60 * MIN)).action, Action::Attempt { v: 4 }, "forgotten after 6 h");
959        assert_eq!(m.status(&at(6 * 60 * MIN)).unwrap().note, None);
960    }
961
962    #[test]
963    fn a_relay_that_took_v2_gets_v2_and_still_a_goodbye() {
964        let mut m = machine();
965        m.next(Event::Start, at(0));
966        assert_eq!(m.next(answer(Answer::Malformed), at(0)).action, Action::Attempt { v: 2 });
967        let step = m.next(Event::Answer { sent_v: 2, answer: Answer::Stored, relay_date_ms: None }, at(0));
968        assert!(step.logs[0].text.contains("as v2"), "{:?}", step.logs);
969        let status = m.status(&at(0)).unwrap();
970        assert_eq!(status.record_version, 2);
971        assert!(status.note.as_deref().unwrap().contains("agent summaries"), "{status:?}");
972        assert_eq!(m.next(Event::Due, at(60 * S)).action, Action::Attempt { v: 2 });
973        assert!(m.farewell(&at(60 * S)).is_some(), "a relay that takes v2 takes the v3 goodbye");
974    }
975
976    #[test]
977    fn the_v1_memory_is_only_set_by_a_fallback_that_worked() {
978        let mut m = machine();
979        m.next(Event::Start, at(0));
980        m.next(answer(Answer::Malformed), at(0));
981        m.next(Event::Answer { sent_v: 2, answer: unreachable_answer(), relay_date_ms: None }, at(0));
982        assert_eq!(m.state(), Some(PresenceState::Retrying));
983        assert_eq!(m.next(Event::Due, at(MIN)).action, Action::Attempt { v: 4 }, "v4 again: v2 never landed");
984    }
985
986    fn unreachable_answer() -> Answer {
987        Answer::Unavailable { reason: "cloud unreachable".into(), detail: String::new() }
988    }
989
990    #[test]
991    fn records_are_stamped_with_the_relays_clock() {
992        let fast = 15 * MIN; // this computer's clock is 15 min fast
993        for (skew_label, local_minus_relay) in [("fast", fast as i64), ("slow", -(fast as i64))] {
994            let mut m = machine();
995            let now = at(0);
996            assert_eq!(m.stamp(now.wall_ms), now.wall_ms, "no answer yet: the local clock");
997            let relay_date = (now.wall_ms as i64 - local_minus_relay) as u64;
998            let step = m.next(Event::Answer { sent_v: 4, answer: Answer::Stored, relay_date_ms: Some(relay_date) }, now);
999            assert!(step.logs.iter().any(|l| l.text.contains("differs from the relay's by 15 min")), "{skew_label}");
1000            assert_eq!(m.stamp(now.wall_ms), relay_date, "{skew_label}");
1001            assert_eq!(m.stamp(now.wall_ms + 5_000), relay_date + 5_000);
1002            let status = m.status(&now).unwrap();
1003            assert_eq!(status.offset_ms, Some(-local_minus_relay));
1004            assert_eq!(status.note.as_deref(), Some("Your clock differs from the cloud's by 15 min"), "{skew_label}");
1005        }
1006    }
1007
1008    #[test]
1009    fn a_small_offset_is_used_but_not_shown() {
1010        let mut m = machine();
1011        let now = at(0);
1012        m.next(Event::Answer { sent_v: 4, answer: Answer::Stored, relay_date_ms: Some(now.wall_ms + 90_000) }, now);
1013        assert_eq!(m.stamp(now.wall_ms), now.wall_ms + 90_000);
1014        assert_eq!(m.status(&now).unwrap().note, None);
1015    }
1016
1017    #[test]
1018    fn an_answer_without_a_date_keeps_the_last_offset() {
1019        let mut m = machine();
1020        let now = at(0);
1021        m.next(Event::Health { version: None, relay_date_ms: Some(now.wall_ms - 1000) }, now);
1022        m.next(answer(Answer::Stored), now);
1023        assert_eq!(m.status(&now).unwrap().offset_ms, Some(-1000));
1024    }
1025
1026    #[test]
1027    fn a_clock_refusal_is_sent_again_at_once_with_the_corrected_time() {
1028        let mut m = machine();
1029        m.next(Event::Start, at(0));
1030        let relay_date = WALL - 15 * MIN;
1031        let step = m.next(
1032            Event::Answer {
1033                sent_v: 4,
1034                answer: Answer::ClockRefused("published_at_ms is too far from the relay's clock".into()),
1035                relay_date_ms: Some(relay_date),
1036            },
1037            at(0),
1038        );
1039        assert_eq!(step.action, Action::Attempt { v: 4 });
1040        assert_eq!(m.stamp(WALL), relay_date);
1041        let step = m.next(
1042            answer(Answer::ClockRefused("published_at_ms is too far from the relay's clock".into())),
1043            at(0),
1044        );
1045        assert_eq!(step.action, Action::Wait, "once only");
1046        assert_eq!(m.state(), Some(PresenceState::Rejected));
1047    }
1048
1049    #[test]
1050    fn a_long_outage_warns_once_and_each_state_change_logs_once() {
1051        let mut m = publishing();
1052        let first = m.next(unreachable(), at(MIN));
1053        assert_eq!(first.logs.len(), 1);
1054        assert!(first.logs[0].text.starts_with("retrying: cloud unreachable"), "{:?}", first.logs);
1055        assert!(!first.logs[0].warn);
1056        let mut warnings = 0;
1057        for i in 2..30u64 {
1058            let step = m.next(unreachable(), at(i * MIN));
1059            assert!(step.logs.iter().all(|l| l.warn), "no line per retry: {:?}", step.logs);
1060            warnings += step.logs.len();
1061            if i == 11 {
1062                assert_eq!(warnings, 1, "warned once 10 min in");
1063                assert_eq!(step.logs[0].text, "not published for 10 min: cloud unreachable");
1064            }
1065        }
1066        assert_eq!(warnings, 1);
1067        // A success ends the outage; the next one warns again.
1068        assert_eq!(m.next(answer(Answer::Stored), at(30 * MIN)).logs.len(), 1);
1069        m.next(unreachable(), at(31 * MIN));
1070        assert!(m.next(unreachable(), at(41 * MIN)).logs.iter().any(|l| l.warn));
1071    }
1072
1073    #[test]
1074    fn the_status_says_when_and_why() {
1075        let mut m = machine();
1076        assert_eq!(m.status(&at(0)), None, "nothing to say before the first try");
1077        m.next(answer(Answer::Stored), at(0));
1078        m.next(unreachable(), at_rand(30 * S, 0.5));
1079        let status = m.status(&at(31 * S)).unwrap();
1080        assert_eq!(
1081            status,
1082            PresenceStatusResult {
1083                state: PresenceState::Retrying,
1084                since_ms: WALL + 30 * S,
1085                last_ok_ms: Some(WALL),
1086                last_error: Some("cloud unreachable".into()),
1087                next_try_ms: Some(WALL + 30 * S + 2500),
1088                relay_version: None,
1089                offset_ms: None,
1090                record_version: 4,
1091                note: None,
1092                off_reason: None,
1093            }
1094        );
1095    }
1096
1097    #[test]
1098    fn statuses_and_errors_are_classified() {
1099        assert_eq!(classify(200, None, None), Answer::Stored);
1100        assert_eq!(classify(204, None, None), Answer::Stored);
1101        assert_eq!(classify(409, Some("a newer record is stored"), None), Answer::Stored, "newest wins: as good as stored");
1102        assert_eq!(classify(404, Some("Not Found"), None), Answer::NoRoute);
1103        assert_eq!(classify(400, Some(MALFORMED_RECORD), None), Answer::Malformed);
1104        assert_eq!(
1105            classify(400, Some("published_at_ms is too far from the relay's clock"), None),
1106            Answer::ClockRefused("published_at_ms is too far from the relay's clock".into())
1107        );
1108        assert_eq!(classify(400, Some("bad sig"), None), Answer::Refused("bad sig".into()));
1109        assert_eq!(classify(410, None, None), Answer::Refused("the relay answered 410".into()));
1110        let ra = Some(Duration::from_secs(7));
1111        assert_eq!(classify(429, None, ra), Answer::TooMany { retry_after: ra });
1112        assert_eq!(
1113            classify(503, Some("down"), None),
1114            Answer::Unavailable { reason: "the cloud answered 503".into(), detail: "down".into() }
1115        );
1116        assert!(matches!(classify(401, None, None), Answer::Unavailable { .. }));
1117    }
1118}
agentmux_srv/muxbus/wan_presence/machine.rs

agentmux_srv/muxbus/wan_presence/
machine.rs
