agentmux_srv/muxbus/
wan_presence.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Publish this install's presence record to the account's relay, so a
5//! signed-in phone off the LAN can list it (agentmux-mobile's
6//! SPEC_FLEET_HOST_TAGS_AND_CLOUD_HOSTS_2026_10_06 §6.3).
7//!
8//! The record (`agentmux_common::install_presence`, version 2) is built from
9//! the LAN fleet feed's snapshot (agent names, kinds and states, the channel
10//! count) plus this install's platform, hostname, channel and version, and
11//! signed with the WAN instance key, the same identity `wan_publish`
12//! certifies agent keys with. It goes to the account the shared MuxBus
13//! sign-in names, with the same token and relay base URL `wan_publish` uses.
14//!
15//! **When.** The rules live in [`machine`], a pure state machine
16//! (`signed_out`, `publishing`, `retrying`, `unsupported`, `rejected`);
17//! [`Driver`] runs it. While publishing: on start, on every change of the
18//! agents, their kinds or the channel count (debounced [`DEBOUNCE`]), on a
19//! change of agent states alone at most every [`STATE_MIN_INTERVAL`]
20//! (agentmux-mobile's SPEC_AGENT_STATUS_AND_LIVE_PANE_FEED_2026_10_07 §3.3,
21//! so busy agents don't publish on every turn), and every minute otherwise.
22//! A failure backs off with jitter; a relay without the route (404) is tried
23//! again within minutes, and at once on a sign-in, a network change, a wake
24//! from sleep or a new relay version. Not signed in: no request at all. One
25//! request at a time, so an older record never lands after a newer one.
26//! Neither the token nor the signature is ever logged.
27//!
28//! **Goodbye.** On a clean quit ([`goodbye_on_shutdown`]), a sign-out
29//! ([`goodbye_before_sign_out`], then [`signed_out`]) and when publishing is
30//! turned off, one last record (v3, `gone`) tells the relay this computer
31//! went offline, so devices stop listing it as live at once rather than
32//! minutes later. Best effort, at most [`GOODBYE_TIMEOUT`]; a relay that
33//! refuses it (an older one) is not asked again, and nothing waits on it.
34//!
35//! **Who publishes** is [`policy`]: not a dev build, a headless srv, an
36//! isolated home or a test harness, unless forced; never when the setting
37//! is off.
38//!
39//! `presence.status` reads [`status`]; `presence.publish-now` calls
40//! [`publish_now`]; a sign-in calls [`sign_in_changed`]; a change of the
41//! settings calls [`settings_changed`].
42
43mod machine;
44pub(crate) mod policy;
45
46use std::future::Future;
47use std::sync::atomic::{AtomicBool, Ordering};
48use std::sync::{Arc, Mutex, OnceLock};
49use std::time::Duration;
50
51use agentmux_common::install_presence::{InstallPresence, PresenceAgent, PRESENCE_VERSION_GOODBYE};
52use tokio::sync::mpsc;
53use tokio_util::sync::CancellationToken;
54
55use crate::backend::fleet_feed::{FleetFeed, FleetSnapshot};
56use crate::backend::rpc_types::{PresenceOffReason, PresenceStatusResult};
57use crate::backend::storage::store::Store;
58use crate::backend::storage::wan_identity::{WanIdentityStore, WanInstance};
59use crate::backend::wconfig::ConfigState;
60use machine::{Action, Answer, Event, Farewell, Machine, Now};
61
62/// A change is published once the agents, kinds and channel count have been
63/// quiet this long.
64const DEBOUNCE: Duration = Duration::from_secs(2);
65
66/// The least time between two publishes when only agent states changed.
67const STATE_MIN_INTERVAL: Duration = Duration::from_secs(10);
68
69const PUBLISH_TIMEOUT_SECS: u64 = 10;
70
71/// The longest a goodbye may take, so a quit or a sign-out is never held up.
72pub(crate) const GOODBYE_TIMEOUT: Duration = Duration::from_secs(2);
73
74/// How often the network addresses and the clock are looked at, for a
75/// network change or a wake from sleep.
76const WATCH_EVERY: Duration = Duration::from_secs(15);
77
78/// A watch tick this much later on the wall clock than it should be means
79/// the computer slept in between.
80const WAKE_SLACK: Duration = Duration::from_secs(30);
81
82const OUT_OF_BOUNDS: &str = "the record is out of bounds (hostname, channel or version)";
83
84/// The record for `snapshot`, as version `v`. An agent whose kind isn't
85/// known is left out: the record requires one, and the feed doesn't guess.
86/// An agent whose state isn't known carries none, and one with no line no line.
87pub(crate) fn build_record(
88    instance: &WanInstance,
89    feed: &FleetFeed,
90    snapshot: &FleetSnapshot,
91    v: u32,
92    published_at_ms: u64,
93) -> Option<InstallPresence> {
94    let agents = snapshot.agents.iter().filter_map(|name| {
95        snapshot.agent_kinds.get(name).map(|kind| PresenceAgent {
96            name: name.clone(),
97            kind: (*kind).to_string(),
98            state: snapshot.agent_status.get(name).map(|s| s.state.as_str().to_string()),
99            line: snapshot.agent_lines.get(name).cloned(),
100        })
101    });
102    InstallPresence::sign_version(
103        v,
104        &instance.private_key,
105        feed.hostname(),
106        feed.channel(),
107        &crate::backend::host_os::sanitize_os(feed.os()).unwrap_or_default(),
108        feed.version(),
109        snapshot.channels_running,
110        agents,
111        published_at_ms,
112    )
113}
114
115/// An HTTP date (`Date`, `Retry-After`) as unix ms.
116pub(crate) fn parse_http_date(value: &str) -> Option<u64> {
117    let at = chrono::DateTime::parse_from_rfc2822(value.trim()).ok()?;
118    u64::try_from(at.timestamp_millis()).ok()
119}
120
121/// `Retry-After`: seconds, or a date read against the relay's own `Date`
122/// (else this computer's clock).
123pub(crate) fn parse_retry_after(value: &str, relay_now_ms: Option<u64>, local_now_ms: u64) -> Option<Duration> {
124    if let Ok(secs) = value.trim().parse::<u64>() {
125        return Some(Duration::from_secs(secs));
126    }
127    let at = parse_http_date(value)?;
128    Some(Duration::from_millis(at.saturating_sub(relay_now_ms.unwrap_or(local_now_ms))))
129}
130
131fn header_date(headers: &reqwest::header::HeaderMap) -> Option<u64> {
132    parse_http_date(headers.get(reqwest::header::DATE)?.to_str().ok()?)
133}
134
135/// What one relay request came to: the answer, and the relay's clock.
136#[derive(Debug, PartialEq, Eq)]
137pub(crate) struct RelayReply {
138    pub answer: Answer,
139    pub date_ms: Option<u64>,
140    /// The HTTP status, when the relay answered at all. A goodbye needs a
141    /// 2xx: [`Answer::Stored`] also covers a 409, where the relay kept a newer
142    /// record and the goodbye did not take effect.
143    pub status: Option<u16>,
144}
145
146/// `PUT /wan-instances/:instance_id/presence`. A pure HTTP operation, like
147/// `wan_publish::publish_record`, so tests can point it at a stub.
148pub(crate) async fn put_presence(
149    base_url: &str,
150    http: &reqwest::Client,
151    token: &str,
152    record: &InstallPresence,
153) -> RelayReply {
154    let url = format!(
155        "{}/wan-instances/{}/presence",
156        base_url.trim_end_matches('/'),
157        record.instance_id
158    );
159    let resp = http
160        .put(&url)
161        .header("Authorization", format!("Bearer {token}"))
162        .timeout(Duration::from_secs(PUBLISH_TIMEOUT_SECS))
163        .json(record)
164        .send()
165        .await;
166    let r = match resp {
167        Ok(r) => r,
168        Err(e) => {
169            return RelayReply {
170                answer: Answer::Unavailable { reason: "cloud unreachable".into(), detail: e.to_string() },
171                date_ms: None,
172                status: None,
173            }
174        }
175    };
176    let status = r.status().as_u16();
177    let date_ms = header_date(r.headers());
178    let retry_after = r
179        .headers()
180        .get(reqwest::header::RETRY_AFTER)
181        .and_then(|v| v.to_str().ok())
182        .and_then(|v| parse_retry_after(v, date_ms, agentmux_common::time::now_ms_u64()));
183    let body = r.text().await.unwrap_or_default();
184    let error = serde_json::from_str::<serde_json::Value>(&body)
185        .ok()
186        .and_then(|v| v.get("error").and_then(|e| e.as_str()).map(str::to_string))
187        .unwrap_or_else(|| body.clone());
188    let error: String = error.chars().take(200).collect();
189    let error = (!error.is_empty()).then_some(error);
190    RelayReply { answer: machine::classify(status, error.as_deref(), retry_after), date_ms, status: Some(status) }
191}
192
193/// The relay's unauthenticated `GET /api/health`: its version, and its clock.
194pub(crate) async fn relay_health(base_url: &str, http: &reqwest::Client) -> (Option<String>, Option<u64>) {
195    #[derive(serde::Deserialize)]
196    struct Health {
197        version: Option<String>,
198    }
199    let url = format!("{}/api/health", base_url.trim_end_matches('/'));
200    match http.get(&url).timeout(Duration::from_secs(PUBLISH_TIMEOUT_SECS)).send().await {
201        Ok(r) => {
202            let date_ms = header_date(r.headers());
203            let version = if r.status().is_success() {
204                r.json::<Health>().await.ok().and_then(|h| h.version).filter(|v| !v.is_empty())
205            } else {
206                None
207            };
208            (version, date_ms)
209        }
210        Err(_) => (None, None),
211    }
212}
213
214/// What a try needs besides the relay: the sign-in, the signed record, and
215/// whether to publish at all.
216pub(crate) trait Session: Sync {
217    /// The MuxBus access token; `None` when not signed in.
218    fn token(&self) -> impl Future<Output = Option<String>> + Send;
219    /// The signed record for `snapshot`, or why there is none.
220    fn record(&self, snapshot: &FleetSnapshot, v: u32, published_at_ms: u64) -> Result<InstallPresence, String>;
221    /// Why this install doesn't publish now; `None` when it does.
222    fn off(&self) -> Option<PresenceOffReason>;
223}
224
225struct LiveSession {
226    wan: Arc<WanIdentityStore>,
227    feed: Arc<FleetFeed>,
228    id_store: Arc<Store>,
229    config: Arc<ConfigState>,
230    install: policy::Install,
231}
232
233impl Session for LiveSession {
234    async fn token(&self) -> Option<String> {
235        // The MuxBus credential is registered with the broker once the cloud
236        // subscriber runs; without it there is no sign-in to use, and asking
237        // the broker to refresh it would only log a warning every minute.
238        let scheduler = crate::broker::get_global()?;
239        scheduler.state(crate::muxbus::CREDENTIAL_ID)?;
240        crate::muxbus::cloud_subscriber::load_valid_token(&self.id_store, &scheduler).await
241    }
242
243    fn record(&self, snapshot: &FleetSnapshot, v: u32, published_at_ms: u64) -> Result<InstallPresence, String> {
244        let instance = self
245            .wan
246            .instance_ensure(&crate::backend::reactive::registry::local_host_label())
247            .map_err(|e| format!("no instance: {e}"))?;
248        build_record(&instance, &self.feed, snapshot, v, published_at_ms).ok_or_else(|| OUT_OF_BOUNDS.to_string())
249    }
250
251    fn off(&self) -> Option<PresenceOffReason> {
252        self.install.off_reason(policy::setting_on(&self.config.get_settings()))
253    }
254}
255
256/// Send the goodbye for `snapshot` (v3, `gone`): whether the relay stored
257/// it, which only a 2xx says (a 409 means it kept a newer, live record). Gives up after `timeout`. A relay that refuses v3 (an older one)
258/// gets nothing more: the record is never sent again as v1 or v2.
259pub(crate) async fn send_goodbye<S: Session>(
260    session: &S,
261    http: &reqwest::Client,
262    base_url: &str,
263    snapshot: &FleetSnapshot,
264    published_at_ms: u64,
265    timeout: Duration,
266) -> bool {
267    let say = async {
268        let token = session.token().await?;
269        let record = session.record(snapshot, PRESENCE_VERSION_GOODBYE, published_at_ms).ok()?;
270        Some(matches!(put_presence(base_url, http, &token, &record).await.status, Some(200..=299)))
271    };
272    matches!(tokio::time::timeout(timeout, say).await, Ok(Some(true)))
273}
274
275/// One try: the record for `snapshot` as version `v`, or nothing without a
276/// sign-in.
277pub(crate) async fn attempt<S: Session>(
278    session: &S,
279    http: &reqwest::Client,
280    base_url: &str,
281    snapshot: &FleetSnapshot,
282    v: u32,
283    published_at_ms: u64,
284) -> Event {
285    let Some(token) = session.token().await else { return Event::NoSignIn };
286    let record = match session.record(snapshot, v, published_at_ms) {
287        Ok(record) => record,
288        Err(why) => return Event::Answer { sent_v: v, answer: Answer::Refused(why), relay_date_ms: None },
289    };
290    let reply = put_presence(base_url, http, &token, &record).await;
291    Event::Answer { sent_v: v, answer: reply.answer, relay_date_ms: reply.date_ms }
292}
293
294/// Uniform in `[0, 1)`, for jitter, from the OS's random source.
295fn random_unit() -> f64 {
296    let bytes = *uuid::Uuid::new_v4().as_bytes();
297    let mut word = [0u8; 8];
298    // The first six bytes of a v4 UUID are all random.
299    word[..6].copy_from_slice(&bytes[..6]);
300    u64::from_le_bytes(word) as f64 / (1u64 << 48) as f64
301}
302
303/// What the driver shares with the rest of srv.
304#[derive(Default)]
305pub(crate) struct Shared {
306    /// For `presence.status`.
307    pub status: Mutex<Option<PresenceStatusResult>>,
308    /// Whether a goodbye is due if this install went away now.
309    pub farewell: Mutex<Option<Farewell>>,
310    /// Set while a goodbye from outside the driver is said (and for good
311    /// at a quit): no regular record may land after it.
312    pub hold: AtomicBool,
313}
314
315impl Shared {
316    const fn new() -> Self {
317        Self { status: Mutex::new(None), farewell: Mutex::new(None), hold: AtomicBool::new(false) }
318    }
319}
320
321/// Runs [`Machine`] against the relay: does what each step says, then waits
322/// for the step's wait, a signal ([`Event::SignIn`], [`Event::PublishNow`],
323/// [`Event::NetworkChanged`], [`Event::Woke`], [`Event::SettingsChanged`],
324/// [`Event::Goodbye`]) or a change of the snapshot. While publishing, a
325/// change is debounced as before; otherwise it only marks the record dirty.
326pub(crate) struct Driver<'a, S> {
327    pub session: &'a S,
328    pub http: &'a reqwest::Client,
329    pub base_url: &'a str,
330    pub config: &'a machine::Config,
331    pub timing: &'a Timing,
332    pub shared: &'a Shared,
333}
334
335impl<S: Session> Driver<'_, S> {
336    pub(crate) async fn run(
337        &self,
338        mut changes: tokio::sync::watch::Receiver<FleetSnapshot>,
339        mut signals: mpsc::UnboundedReceiver<Event>,
340        cancel: &CancellationToken,
341    ) {
342        let started = tokio::time::Instant::now();
343        let now = || Now {
344            mono_ms: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
345            wall_ms: agentmux_common::time::now_ms_u64(),
346            rand: random_unit(),
347        };
348        let mut machine = Machine::new(self.config.clone(), self.base_url);
349        let mut sent = changes.borrow_and_update().clone();
350        let mut sent_at = tokio::time::Instant::now();
351        let mut event = match self.session.off() {
352            None => Event::Start,
353            off => Event::Policy { off },
354        };
355        loop {
356            let at = now();
357            let step = machine.next(event, at);
358            for log in &step.logs {
359                if log.warn {
360                    tracing::warn!("wan presence: {}", log.text);
361                } else {
362                    tracing::info!("wan presence: {}", log.text);
363                }
364            }
365            if let (Some(status), Ok(mut slot)) = (machine.status(&at), self.shared.status.lock()) {
366                *slot = Some(status);
367            }
368            if let Ok(mut slot) = self.shared.farewell.lock() {
369                *slot = machine.farewell(&at);
370            }
371            event = match step.action {
372                Action::Attempt { .. } if self.shared.hold.load(Ordering::SeqCst) => Event::NoSignIn,
373                Action::Attempt { v } => {
374                    sent = changes.borrow_and_update().clone();
375                    sent_at = tokio::time::Instant::now();
376                    let published_at_ms = machine.stamp(agentmux_common::time::now_ms_u64());
377                    attempt(self.session, self.http, self.base_url, &sent, v, published_at_ms).await
378                }
379                Action::Goodbye { published_at_ms } => {
380                    let snapshot = changes.borrow().clone();
381                    let stored =
382                        send_goodbye(self.session, self.http, self.base_url, &snapshot, published_at_ms, GOODBYE_TIMEOUT)
383                            .await;
384                    Event::Goodbye { stored }
385                }
386                Action::CheckHealth => {
387                    let (version, relay_date_ms) = relay_health(self.base_url, self.http).await;
388                    Event::Health { version, relay_date_ms }
389                }
390                Action::Wait => {
391                    let follow = machine.follows_changes();
392                    tokio::select! {
393                        _ = cancel.cancelled() => return,
394                        _ = tokio::time::sleep(step.wait) => Event::Due,
395                        Some(signal) = signals.recv() => match signal {
396                            Event::SettingsChanged => Event::Policy { off: self.session.off() },
397                            signal => signal,
398                        },
399                        changed = changes.changed() => {
400                            if changed.is_err() {
401                                return;
402                            }
403                            if follow && !wait_to_publish(&mut changes, &sent, sent_at, cancel, self.timing).await {
404                                return;
405                            }
406                            Event::Changed
407                        }
408                    }
409                }
410            };
411        }
412    }
413}
414
415static SIGNALS: OnceLock<mpsc::UnboundedSender<Event>> = OnceLock::new();
416static SHARED: Shared = Shared::new();
417
418/// The running publisher's session and relay, for a goodbye said from
419/// outside the driver.
420struct Publisher {
421    session: LiveSession,
422    http: reqwest::Client,
423    base_url: String,
424}
425
426static PUBLISHER: OnceLock<Publisher> = OnceLock::new();
427
428/// The publisher's status; `None` until it has one (or without a publisher).
429pub(crate) fn status() -> Option<PresenceStatusResult> {
430    SHARED.status.lock().ok()?.clone()
431}
432
433/// The settings changed: whether to publish is read again.
434pub(crate) fn settings_changed() {
435    if let Some(tx) = SIGNALS.get() {
436        let _ = tx.send(Event::SettingsChanged);
437    }
438}
439
440/// Hold the publisher and say goodbye if one is due: whether the relay
441/// stored it. At most [`GOODBYE_TIMEOUT`].
442pub(crate) async fn farewell_from(shared: &Shared, say: impl AsyncFnOnce(Farewell) -> bool) -> bool {
443    shared.hold.store(true, Ordering::SeqCst);
444    let due = shared.farewell.lock().ok().and_then(|f| *f);
445    match due {
446        Some(farewell) => say(farewell).await,
447        None => false,
448    }
449}
450
451async fn farewell_now() -> bool {
452    let Some(p) = PUBLISHER.get() else { return false };
453    farewell_from(&SHARED, async |farewell: Farewell| {
454        let snapshot = p.session.feed.snapshot();
455        let at = farewell.stamp(agentmux_common::time::now_ms_u64());
456        send_goodbye(&p.session, &p.http, &p.base_url, &snapshot, at, GOODBYE_TIMEOUT).await
457    })
458    .await
459}
460
461/// A clean quit: say goodbye, at most [`GOODBYE_TIMEOUT`]. Nothing is
462/// published after it.
463pub(crate) async fn goodbye_on_shutdown() {
464    if farewell_now().await {
465        tracing::info!("wan presence: said goodbye, devices show this computer as offline");
466    }
467}
468
469/// A sign-out, before the sign-in is cleared (the goodbye needs it): say
470/// goodbye, at most [`GOODBYE_TIMEOUT`], and publish nothing more until
471/// [`signed_out`]. Whether the relay stored it.
472pub(crate) async fn goodbye_before_sign_out() -> bool {
473    farewell_now().await
474}
475
476/// The sign-in is cleared (or clearing it failed): publishing may go on,
477/// and the status says "signed off" when the goodbye was stored.
478pub(crate) fn signed_out(stored: bool) {
479    SHARED.hold.store(false, Ordering::SeqCst);
480    if let Some(tx) = SIGNALS.get() {
481        let _ = tx.send(Event::Goodbye { stored });
482    }
483}
484
485/// Forget the backoff and try at once. False when no publisher runs.
486pub(crate) fn publish_now() -> bool {
487    SIGNALS.get().is_some_and(|tx| tx.send(Event::PublishNow).is_ok())
488}
489
490/// A sign-in, sign-out or account change in this process: try at once.
491pub(crate) fn sign_in_changed() {
492    if let Some(tx) = SIGNALS.get() {
493        let _ = tx.send(Event::SignIn);
494    }
495}
496
497/// Start the publisher. No-op without a WAN identity store (nothing to sign
498/// with). `headless` is srv's `--headless` ([`policy`]). Stops with `token`
499/// (srv shutdown).
500pub(crate) fn spawn(
501    feed: Arc<FleetFeed>,
502    id_store: Arc<Store>,
503    config: Arc<ConfigState>,
504    headless: bool,
505    token: CancellationToken,
506) {
507    let Some(wan) = crate::backend::storage::wan_identity::global() else { return };
508    let install = policy::Install::from_env(headless);
509    let publisher = Publisher {
510        session: LiveSession { wan, feed, id_store, config, install },
511        http: reqwest::Client::new(),
512        base_url: crate::muxbus::relay::rest_base_url(),
513    };
514    if PUBLISHER.set(publisher).is_err() {
515        return;
516    }
517    let (tx, rx) = mpsc::unbounded_channel();
518    if SIGNALS.set(tx.clone()).is_err() {
519        return;
520    }
521    tokio::spawn(watch_network_and_sleep(tx, token.clone()));
522    tokio::spawn(async move {
523        let Some(p) = PUBLISHER.get() else { return };
524        let changes = p.session.feed.subscribe();
525        let driver = Driver {
526            session: &p.session,
527            http: &p.http,
528            base_url: &p.base_url,
529            config: &machine::LIVE,
530            timing: &LIVE_TIMING,
531            shared: &SHARED,
532        };
533        driver.run(changes, rx, &token).await;
534    });
535}
536
537/// Whether the computer slept between two watch ticks `tick` apart: the
538/// wall clock moved much further than the monotonic clock (which stops
539/// during sleep on Linux and macOS), or much further than the tick itself
540/// (where the monotonic clock runs on through sleep).
541pub(crate) fn woke(tick: Duration, mono_elapsed: Duration, wall_elapsed_ms: i64) -> bool {
542    let Ok(wall) = u64::try_from(wall_elapsed_ms) else { return false };
543    let wall = Duration::from_millis(wall);
544    wall > mono_elapsed + WAKE_SLACK || wall > tick + WAKE_SLACK
545}
546
547/// This computer's addresses, loopback aside, sorted; `None` when they
548/// can't be read.
549fn network_addresses() -> Option<Vec<std::net::IpAddr>> {
550    let mut ips: Vec<_> = if_addrs::get_if_addrs()
551        .ok()?
552        .into_iter()
553        .filter(|i| !i.is_loopback())
554        .map(|i| i.ip())
555        .collect();
556    ips.sort();
557    ips.dedup();
558    Some(ips)
559}
560
561/// Whether the addresses moved; an unreadable list is no change.
562pub(crate) fn network_changed(before: &Option<Vec<std::net::IpAddr>>, after: &Option<Vec<std::net::IpAddr>>) -> bool {
563    matches!((before, after), (Some(a), Some(b)) if a != b)
564}
565
566/// Every [`WATCH_EVERY`], tell the publisher about a wake from sleep or a
567/// change of network addresses. Cheap: a clock read and the interface list.
568async fn watch_network_and_sleep(signals: mpsc::UnboundedSender<Event>, cancel: CancellationToken) {
569    let mut addresses = network_addresses();
570    let mut last = (tokio::time::Instant::now(), agentmux_common::time::now_ms_u64());
571    loop {
572        tokio::select! {
573            _ = cancel.cancelled() => return,
574            _ = tokio::time::sleep(WATCH_EVERY) => {}
575        }
576        let here = (tokio::time::Instant::now(), agentmux_common::time::now_ms_u64());
577        let slept = woke(WATCH_EVERY, here.0 - last.0, here.1 as i64 - last.1 as i64);
578        last = here;
579        let current = network_addresses();
580        let moved = network_changed(&addresses, &current);
581        if current.is_some() {
582            addresses = current;
583        }
584        let event = if slept {
585            Some(Event::Woke)
586        } else if moved {
587            Some(Event::NetworkChanged)
588        } else {
589            None
590        };
591        if let Some(event) = event {
592            if signals.send(event).is_err() {
593                return;
594            }
595        }
596    }
597}
598
599/// The publisher's waits, apart so tests can shorten them.
600pub(crate) struct Timing {
601    debounce: Duration,
602    state_min_interval: Duration,
603    max_settle: Duration,
604}
605
606const LIVE_TIMING: Timing = Timing {
607    debounce: DEBOUNCE,
608    state_min_interval: STATE_MIN_INTERVAL,
609    max_settle: machine::LIVE.interval,
610};
611
612/// Whether anything but agent states and lines differs: the agents, their
613/// kinds or the channel count.
614pub(crate) fn structure_changed(a: &FleetSnapshot, b: &FleetSnapshot) -> bool {
615    a.agents != b.agents || a.agent_kinds != b.agent_kinds || a.channels_running != b.channels_running
616}
617
618/// After a change of the snapshot, wait until it is time to publish: a
619/// change of the agents, kinds or channel count once those have settled
620/// ([`settle`]); a change of agent states or lines alone once
621/// [`STATE_MIN_INTERVAL`] has passed since `sent_at`. A structural change
622/// during that wait switches to settling. `false` on shutdown.
623async fn wait_to_publish(
624    changes: &mut tokio::sync::watch::Receiver<FleetSnapshot>,
625    sent: &FleetSnapshot,
626    sent_at: tokio::time::Instant,
627    token: &CancellationToken,
628    timing: &Timing,
629) -> bool {
630    let due = sent_at + timing.state_min_interval;
631    loop {
632        let structural = structure_changed(sent, &changes.borrow_and_update());
633        if structural {
634            return settle(changes, token, timing).await;
635        }
636        tokio::select! {
637            _ = token.cancelled() => return false,
638            _ = tokio::time::sleep_until(due) => return true,
639            changed = changes.changed() => {
640                if changed.is_err() {
641                    return false;
642                }
643            }
644        }
645    }
646}
647
648/// Wait until the agents, kinds and channel count have been quiet for
649/// [`DEBOUNCE`], at most a publish interval in all. A change of states alone
650/// does not restart the wait: busy agents would otherwise hold a new agent
651/// back for the full interval. `false` on shutdown.
652async fn settle(
653    changes: &mut tokio::sync::watch::Receiver<FleetSnapshot>,
654    token: &CancellationToken,
655    timing: &Timing,
656) -> bool {
657    let deadline = tokio::time::Instant::now() + timing.max_settle;
658    let mut reference = changes.borrow_and_update().clone();
659    let mut quiet_at = tokio::time::Instant::now() + timing.debounce;
660    loop {
661        tokio::select! {
662            _ = token.cancelled() => return false,
663            _ = tokio::time::sleep_until(quiet_at) => return true,
664            _ = tokio::time::sleep_until(deadline) => return true,
665            changed = changes.changed() => {
666                if changed.is_err() {
667                    return false;
668                }
669                let current = changes.borrow_and_update().clone();
670                if structure_changed(&reference, &current) {
671                    reference = current;
672                    quiet_at = tokio::time::Instant::now() + timing.debounce;
673                }
674            }
675        }
676    }
677}
678
679#[cfg(test)]
680mod tests;
agentmux_srv/muxbus/wan_presence.rs

agentmux_srv/muxbus/
wan_presence.rs
