agentmux_srv/backend/
fleet_feed.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The names-only fleet feed LAN peers read: `GET /agentmux/fleet` and its
5//! Server-Sent Events stream `GET /agentmux/fleet/events`
6//! (`docs/specs/SPEC_LAN_FLEET_FEED_2026_10_03.md` §2, which links the
7//! mobile spec that owns the wire contract).
8//!
9//! One task compares this instance's reachable agent names once a second and
10//! publishes `(rev, names)` through a `watch`, so the snapshot route and every
11//! stream read the same value and no stream can fall behind: a slow reader
12//! skips straight to the latest state. Agent names only, the same line
13//! `handle_reactive_agent_names` holds for a `lan_key` holder, plus (owner
14//! decision 2026-10-06, agentmux-mobile's
15//! SPEC_FLEET_HOST_TAGS_AND_CLOUD_HOSTS_2026_10_06 §5.1) where each agent
16//! runs (`agent_kinds`), how many channels this machine runs, its platform
17//! and this install's id; and (owner decision D1, 2026-10-07, agentmux-mobile's
18//! SPEC_AGENT_STATUS_AND_LIVE_PANE_FEED_2026_10_07 §3.2, §13.1) each agent's
19//! state and since when (`agent_status`, `backend::agent_state`), with this
20//! machine's clock (`now_ms`); and, while it is up, the viewer listener's port
21//! (same spec §13.2), which opens nothing by itself; and (owner decision
22//! 2026-10-13) each agent's summary line (`agent_lines`, `backend::agent_line`),
23//! absent while `swarm:sharesummaries` is off. The cloud presence publisher
24//! (`muxbus::wan_presence`) reads the same snapshot.
25
26use std::collections::BTreeMap;
27use std::convert::Infallible;
28use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering};
29use std::sync::Arc;
30use std::time::Duration;
31
32use axum::body::Bytes;
33use futures_util::Stream;
34use serde::Serialize;
35use tokio::sync::watch;
36
37use crate::backend::agent_state::AgentStatus;
38
39/// How often the name set is compared.
40pub const CHANGE_POLL_INTERVAL: Duration = Duration::from_secs(1);
41/// Comment line sent on an idle stream, so a client can tell a dead TCP
42/// connection (30 s of silence, spec §6) from a quiet fleet.
43pub const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
44/// Concurrent event streams per srv; beyond this a request gets 503, so a LAN
45/// peer holding the `lan_key` cannot exhaust the srv.
46pub const MAX_STREAMS: usize = 32;
47/// The stream's first write: how long an SSE client waits before reconnecting.
48const RETRY_PREAMBLE: &[u8] = b"retry: 3000\n\n";
49const HEARTBEAT: &[u8] = b": hb\n\n";
50
51/// An agent's kind as the feed reports it: `"host"` or `"container"`
52/// (`operator_config_seed::agent_kind`).
53pub type AgentKind = &'static str;
54
55/// One state of the fleet. `rev` starts at 1 and goes up by one on every
56/// change of any field below; `agents` is sorted case-insensitively and
57/// de-duplicated; `agent_kinds` and `agent_status` have an entry for each of
58/// `agents` whose kind or state is known.
59#[derive(Debug, Clone, PartialEq, Eq)]
60pub struct FleetSnapshot {
61    pub rev: u64,
62    pub agents: Arc<Vec<String>>,
63    pub agent_kinds: Arc<BTreeMap<String, AgentKind>>,
64    /// This channel plus the other running channels of this machine, 1..=99.
65    pub channels_running: u32,
66    pub agent_status: Arc<BTreeMap<String, AgentStatus>>,
67    /// Each agent's summary line, by name, for agents that have one.
68    pub agent_lines: Arc<BTreeMap<String, String>>,
69    /// The viewer listener's port while it is up (`backend::viewer`), so a
70    /// paired device finds it again after an address change.
71    pub viewer_port: Option<u16>,
72}
73
74/// What one look at the instance found: each reachable agent with its kind,
75/// `None` when its block couldn't be read, the channel count, and the state
76/// of each agent that has one, by name.
77#[derive(Debug, Clone, Default)]
78pub struct FleetObservation {
79    pub agents: Vec<(String, Option<AgentKind>)>,
80    pub channels_running: u32,
81    pub agent_status: BTreeMap<String, AgentStatus>,
82    pub agent_lines: BTreeMap<String, String>,
83}
84
85/// The `/agentmux/fleet` body, in the contract's field order.
86#[derive(Serialize)]
87struct FleetBody<'a> {
88    epoch: &'a str,
89    rev: u64,
90    hostname: &'a str,
91    channel: &'a str,
92    version: &'a str,
93    agents: &'a [String],
94    os: &'a str,
95    #[serde(skip_serializing_if = "Option::is_none")]
96    install_id: Option<&'a str>,
97    channels_running: u32,
98    agent_kinds: &'a BTreeMap<String, AgentKind>,
99    now_ms: u64,
100    agent_status: &'a BTreeMap<String, AgentStatus>,
101    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
102    agent_lines: &'a BTreeMap<String, String>,
103    #[serde(skip_serializing_if = "Option::is_none")]
104    viewer_port: Option<u16>,
105}
106
107pub struct FleetFeed {
108    /// 16 lowercase hex chars, new on every srv launch: a client that sees a
109    /// different epoch knows the instance restarted and takes the snapshot
110    /// as-is rather than comparing `rev`s.
111    epoch: String,
112    hostname: String,
113    channel: String,
114    version: String,
115    os: String,
116    /// This install's WAN instance id; `None` without a WAN identity store.
117    install_id: Option<String>,
118    tx: watch::Sender<FleetSnapshot>,
119    /// Where the viewer port is read from on each look; 0 = not up.
120    viewer_port: Arc<AtomicU16>,
121    streams: Arc<AtomicUsize>,
122    heartbeat: Duration,
123}
124
125/// [`normalize_agents`] for names alone.
126#[cfg(test)]
127pub fn normalize_names(names: impl IntoIterator<Item = String>) -> Vec<String> {
128    normalize_agents(names.into_iter().map(|n| (n, None))).0
129}
130
131/// Sort case-insensitively and drop names that differ only by case
132/// (registration keys are lower-cased, so those are one agent). Ties break on
133/// the exact string so the result does not depend on input order. Each
134/// surviving name keeps its own kind.
135fn normalize_agents(
136    agents: impl IntoIterator<Item = (String, Option<AgentKind>)>,
137) -> (Vec<String>, BTreeMap<String, AgentKind>) {
138    let mut keyed: Vec<(String, String, Option<AgentKind>)> = agents
139        .into_iter()
140        .map(|(n, kind)| (n.to_lowercase(), n, kind))
141        .collect();
142    keyed.sort();
143    keyed.dedup_by(|later, earlier| later.0 == earlier.0);
144    let kinds = keyed
145        .iter()
146        .filter_map(|(_, n, kind)| kind.map(|k| (n.clone(), k)))
147        .collect();
148    (keyed.into_iter().map(|(_, n, _)| n).collect(), kinds)
149}
150
151impl FleetFeed {
152    pub fn new(hostname: String, channel: String, version: String) -> Self {
153        let epoch = uuid::Uuid::new_v4().simple().to_string()[..16].to_string();
154        let (tx, _) = watch::channel(FleetSnapshot {
155            rev: 1,
156            agents: Arc::new(Vec::new()),
157            agent_kinds: Arc::new(BTreeMap::new()),
158            channels_running: 1,
159            agent_status: Arc::new(BTreeMap::new()),
160            agent_lines: Arc::new(BTreeMap::new()),
161            viewer_port: None,
162        });
163        Self {
164            epoch,
165            hostname,
166            channel,
167            version,
168            os: crate::backend::host_os::local_os(),
169            install_id: None,
170            tx,
171            viewer_port: Arc::new(AtomicU16::new(0)),
172            streams: Arc::new(AtomicUsize::new(0)),
173            heartbeat: HEARTBEAT_INTERVAL,
174        }
175    }
176
177    /// Read the viewer port from `port` (`ViewerService::advertised_port_handle`).
178    pub fn with_viewer_port(mut self, port: Arc<AtomicU16>) -> Self {
179        self.viewer_port = port;
180        self
181    }
182
183    /// Set the install id the body reports (`fleet_source::install_id`).
184    pub fn with_install_id(mut self, install_id: Option<String>) -> Self {
185        self.install_id = install_id;
186        self
187    }
188
189    #[cfg(test)]
190    pub fn epoch(&self) -> &str {
191        &self.epoch
192    }
193
194    pub fn hostname(&self) -> &str {
195        &self.hostname
196    }
197
198    pub fn channel(&self) -> &str {
199        &self.channel
200    }
201
202    pub fn version(&self) -> &str {
203        &self.version
204    }
205
206    pub fn os(&self) -> &str {
207        &self.os
208    }
209
210    pub fn snapshot(&self) -> FleetSnapshot {
211        self.tx.borrow().clone()
212    }
213
214    /// A receiver that wakes on every change, for the presence publisher.
215    pub fn subscribe(&self) -> watch::Receiver<FleetSnapshot> {
216        self.tx.subscribe()
217    }
218
219    /// `<epoch>:<rev>`: the SSE `id`, and (quoted) the snapshot's `ETag`.
220    pub fn event_id(&self, snapshot: &FleetSnapshot) -> String {
221        format!("{}:{}", self.epoch, snapshot.rev)
222    }
223
224    /// The `/agentmux/fleet` body, on one line, with `now_ms` read as it is
225    /// served, so a peer can tell how long a state has held without trusting
226    /// its own clock against this one. `now_ms` is not part of the snapshot:
227    /// it never bumps `rev` or changes the `ETag`.
228    pub fn body_json(&self, snapshot: &FleetSnapshot) -> String {
229        self.body_json_at(snapshot, agentmux_common::time::now_ms_u64())
230    }
231
232    fn body_json_at(&self, snapshot: &FleetSnapshot, now_ms: u64) -> String {
233        serde_json::to_string(&FleetBody {
234            epoch: &self.epoch,
235            rev: snapshot.rev,
236            hostname: &self.hostname,
237            channel: &self.channel,
238            version: &self.version,
239            agents: &snapshot.agents,
240            os: &self.os,
241            install_id: self.install_id.as_deref(),
242            channels_running: snapshot.channels_running,
243            agent_kinds: &snapshot.agent_kinds,
244            now_ms,
245            agent_status: &snapshot.agent_status,
246            agent_lines: &snapshot.agent_lines,
247            viewer_port: snapshot.viewer_port,
248        })
249        .unwrap_or_default()
250    }
251
252    /// Record the current fleet; bumps `rev` and wakes every stream only if
253    /// it differs from the last one (names, kinds, channel count, any
254    /// agent's state or line, or the viewer port). Returns whether it did.
255    pub fn observe_fleet(&self, observation: FleetObservation) -> bool {
256        let (names, kinds) = normalize_agents(observation.agents);
257        let channels_running = observation.channels_running.clamp(1, 99);
258        let mut status = observation.agent_status;
259        status.retain(|name, _| names.contains(name));
260        let mut lines = observation.agent_lines;
261        lines.retain(|name, _| names.contains(name));
262        let viewer_port = Some(self.viewer_port.load(Ordering::Relaxed)).filter(|p| *p != 0);
263        self.tx.send_if_modified(|current| {
264            if *current.agents == names
265                && *current.agent_kinds == kinds
266                && current.channels_running == channels_running
267                && *current.agent_status == status
268                && *current.agent_lines == lines
269                && current.viewer_port == viewer_port
270            {
271                return false;
272            }
273            current.rev += 1;
274            current.agents = Arc::new(names);
275            current.agent_kinds = Arc::new(kinds);
276            current.channels_running = channels_running;
277            current.agent_status = Arc::new(status);
278            current.agent_lines = Arc::new(lines);
279            current.viewer_port = viewer_port;
280            true
281        })
282    }
283
284    /// [`Self::observe_fleet`] with names only: no kinds, the channel count
285    /// unchanged.
286    #[cfg(test)]
287    pub fn observe(&self, names: impl IntoIterator<Item = String>) -> bool {
288        let channels_running = self.snapshot().channels_running;
289        self.observe_fleet(FleetObservation {
290            agents: names.into_iter().map(|n| (n, None)).collect(),
291            channels_running,
292            agent_status: BTreeMap::new(),
293            agent_lines: BTreeMap::new(),
294        })
295    }
296
297    /// Start the change-detection task: `observe` is read now, then every
298    /// [`CHANGE_POLL_INTERVAL`] on a blocking thread (it reads the store and
299    /// the shared registry) until `token` is cancelled (srv shutdown), the
300    /// same stop signal the WAL checkpoint loop watches.
301    pub fn spawn_change_detection<F>(
302        self: Arc<Self>,
303        observe: F,
304        token: tokio_util::sync::CancellationToken,
305    ) where
306        F: Fn() -> FleetObservation + Send + Sync + 'static,
307    {
308        let observe = Arc::new(observe);
309        self.observe_fleet(observe());
310        tokio::spawn(async move {
311            let mut tick = tokio::time::interval(CHANGE_POLL_INTERVAL);
312            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
313            loop {
314                tokio::select! {
315                    _ = tick.tick() => {
316                        let observe = Arc::clone(&observe);
317                        if let Ok(observation) = tokio::task::spawn_blocking(move || observe()).await {
318                            self.observe_fleet(observation);
319                        }
320                    }
321                    _ = token.cancelled() => break,
322                }
323            }
324        });
325    }
326
327    /// Open one event stream, or `None` when [`MAX_STREAMS`] are already open.
328    ///
329    /// The slot is held by a guard moved into the stream, so it is released
330    /// whenever the stream is dropped: when hyper drops the response body
331    /// after the client goes away (noticed at the latest on the next
332    /// heartbeat write), or if the response is never sent at all.
333    ///
334    /// With `last_event_id` equal to the current `<epoch>:<rev>` the client is
335    /// already up to date and gets no initial event, only later ones.
336    pub fn open_stream(
337        self: &Arc<Self>,
338        last_event_id: Option<&str>,
339    ) -> Option<impl Stream<Item = Result<Bytes, Infallible>> + Send + 'static> {
340        let slot = StreamSlot::acquire(&self.streams)?;
341        let feed = Arc::clone(self);
342        let mut rx = self.tx.subscribe();
343        let resume_from = last_event_id.map(str::to_string);
344        Some(async_stream::stream! {
345            let _slot = slot;
346            yield Ok(Bytes::from_static(RETRY_PREAMBLE));
347            let current = rx.borrow_and_update().clone();
348            if resume_from.as_deref() != Some(feed.event_id(&current).as_str()) {
349                yield Ok(feed.event_bytes(&current));
350            }
351            let mut heartbeat = tokio::time::interval_at(tokio::time::Instant::now() + feed.heartbeat, feed.heartbeat);
352            heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
353            loop {
354                tokio::select! {
355                    changed = rx.changed() => {
356                        if changed.is_err() {
357                            break;
358                        }
359                        let snapshot = rx.borrow_and_update().clone();
360                        yield Ok(feed.event_bytes(&snapshot));
361                    }
362                    _ = heartbeat.tick() => {
363                        yield Ok(Bytes::from_static(HEARTBEAT));
364                    }
365                }
366            }
367        })
368    }
369
370    /// One `fleet` event. `body_json` is serde output, so it holds no raw
371    /// newline that could break the `data:` line.
372    fn event_bytes(&self, snapshot: &FleetSnapshot) -> Bytes {
373        Bytes::from(format!(
374            "event: fleet\nid: {}\ndata: {}\n\n",
375            self.event_id(snapshot),
376            self.body_json(snapshot)
377        ))
378    }
379
380    #[cfg(test)]
381    pub fn open_streams(&self) -> usize {
382        self.streams.load(Ordering::SeqCst)
383    }
384
385    #[cfg(test)]
386    pub fn with_heartbeat(mut self, every: Duration) -> Self {
387        self.heartbeat = every;
388        self
389    }
390}
391
392/// A held stream slot; releases it on drop.
393struct StreamSlot(Arc<AtomicUsize>);
394
395impl StreamSlot {
396    fn acquire(counter: &Arc<AtomicUsize>) -> Option<Self> {
397        counter
398            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
399                (n < MAX_STREAMS).then_some(n + 1)
400            })
401            .ok()
402            .map(|_| Self(Arc::clone(counter)))
403    }
404}
405
406impl Drop for StreamSlot {
407    fn drop(&mut self) {
408        self.0.fetch_sub(1, Ordering::SeqCst);
409    }
410}
411
412#[cfg(test)]
413mod tests {
414    use super::*;
415    use crate::backend::agent_state::AgentState;
416    use futures_util::StreamExt;
417
418    fn feed() -> Arc<FleetFeed> {
419        Arc::new(FleetFeed::new(
420            "narko".into(),
421            "stable".into(),
422            "0.59.7".into(),
423        ))
424    }
425
426    fn names(list: &[&str]) -> Vec<String> {
427        list.iter().map(|s| s.to_string()).collect()
428    }
429
430    async fn next_chunk(
431        stream: &mut (impl Stream<Item = Result<Bytes, Infallible>> + Unpin),
432    ) -> String {
433        let chunk = tokio::time::timeout(Duration::from_secs(5), stream.next())
434            .await
435            .expect("a chunk within 5s")
436            .expect("stream open")
437            .unwrap();
438        String::from_utf8(chunk.to_vec()).unwrap()
439    }
440
441    #[test]
442    fn names_are_sorted_case_insensitively_and_deduplicated() {
443        assert_eq!(
444            normalize_names(names(&["clamk", "AgentY", "agenty", "Bravo", "alpha"])),
445            names(&["AgentY", "alpha", "Bravo", "clamk"])
446        );
447        // Independent of input order.
448        assert_eq!(
449            normalize_names(names(&["agenty", "AgentY"])),
450            normalize_names(names(&["AgentY", "agenty"]))
451        );
452    }
453
454    #[test]
455    fn the_epoch_is_sixteen_lowercase_hex_chars_and_new_per_feed() {
456        let a = feed();
457        assert_eq!(a.epoch().len(), 16);
458        assert!(a
459            .epoch()
460            .chars()
461            .all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)));
462        assert_ne!(a.epoch(), feed().epoch());
463    }
464
465    #[test]
466    fn rev_starts_at_one_and_moves_only_when_the_name_set_changes() {
467        let f = feed();
468        assert_eq!(f.snapshot().rev, 1);
469        assert!(f.observe(names(&["b", "a"])));
470        assert_eq!(f.snapshot().rev, 2);
471        assert!(!f.observe(names(&["a", "b"])), "same set, other order");
472        assert!(
473            !f.observe(names(&["a", "a", "b"])),
474            "a duplicate is not a change"
475        );
476        assert_eq!(f.snapshot().rev, 2);
477        assert!(f.observe(names(&["a"])));
478        assert_eq!(f.snapshot().rev, 3);
479        assert_eq!(*f.snapshot().agents, names(&["a"]));
480    }
481
482    fn keys_of(body: &str) -> Vec<String> {
483        let v: serde_json::Value = serde_json::from_str(body).unwrap();
484        let mut keys: Vec<_> = v.as_object().unwrap().keys().cloned().collect();
485        keys.sort();
486        keys
487    }
488
489    fn observation(agents: &[(&str, Option<AgentKind>)], channels_running: u32) -> FleetObservation {
490        FleetObservation {
491            agents: agents.iter().map(|(n, k)| (n.to_string(), *k)).collect(),
492            channels_running,
493            ..Default::default()
494        }
495    }
496
497    fn status(state: AgentState, since_ms: u64) -> AgentStatus {
498        AgentStatus { state, since_ms }
499    }
500
501    fn with_status(agents: &[&str], status: &[(&str, AgentStatus)]) -> FleetObservation {
502        FleetObservation {
503            agents: agents.iter().map(|n| (n.to_string(), Some("host"))).collect(),
504            channels_running: 1,
505            agent_status: status.iter().map(|(n, s)| (n.to_string(), *s)).collect(),
506            ..Default::default()
507        }
508    }
509
510    #[test]
511    fn the_body_has_exactly_the_contract_fields() {
512        let f = feed();
513        f.observe_fleet(observation(&[("Clamk", Some("container")), ("AgentY", Some("host"))], 3));
514        let snap = f.snapshot();
515        let body = f.body_json(&snap);
516        assert!(!body.contains('\n'));
517        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
518        let mut want = names(&[
519            "epoch",
520            "rev",
521            "hostname",
522            "channel",
523            "version",
524            "agents",
525            "os",
526            "channels_running",
527            "agent_kinds",
528            "now_ms",
529            "agent_status",
530        ]);
531        want.sort();
532        assert_eq!(keys_of(&body), want, "no install_id without a WAN identity");
533        assert_eq!(v["epoch"], f.epoch());
534        assert_eq!(v["rev"], 2);
535        assert_eq!(v["hostname"], "narko");
536        assert_eq!(v["channel"], "stable");
537        assert_eq!(v["version"], "0.59.7");
538        assert_eq!(v["agents"], serde_json::json!(["AgentY", "Clamk"]));
539        assert_eq!(v["os"], crate::backend::host_os::local_os());
540        assert_eq!(v["channels_running"], 3);
541        assert_eq!(v["agent_kinds"], serde_json::json!({"AgentY": "host", "Clamk": "container"}));
542        assert_eq!(v["agent_status"], serde_json::json!({}));
543        assert!(v["now_ms"].as_u64().unwrap() > 1_700_000_000_000, "this machine's clock, in ms");
544    }
545
546    #[test]
547    fn agent_status_carries_each_known_state_and_leaves_the_rest_out() {
548        let f = feed();
549        f.observe_fleet(with_status(
550            &["AgentY", "Camper", "Clamk"],
551            &[
552                ("AgentY", status(AgentState::Working, 1_000)),
553                ("Camper", status(AgentState::Waiting, 2_000)),
554                ("Gone", status(AgentState::Idle, 3_000)),
555            ],
556        ));
557        let v: serde_json::Value = serde_json::from_str(&f.body_json_at(&f.snapshot(), 5_000)).unwrap();
558        assert_eq!(v["now_ms"], 5_000);
559        assert_eq!(
560            v["agent_status"],
561            serde_json::json!({
562                "AgentY": {"state": "working", "since_ms": 1_000},
563                "Camper": {"state": "waiting", "since_ms": 2_000},
564            }),
565            "Clamk has no state; Gone is not listed"
566        );
567    }
568
569    #[test]
570    fn agent_lines_are_carried_for_listed_agents_and_a_change_bumps_rev() {
571        let f = feed();
572        let with_lines = |lines: &[(&str, &str)]| FleetObservation {
573            agent_lines: lines.iter().map(|(n, l)| (n.to_string(), l.to_string())).collect(),
574            ..with_status(&["AgentY", "Camper"], &[])
575        };
576        assert!(f.observe_fleet(with_lines(&[("AgentY", "Fixing the login race"), ("Gone", "x")])));
577        let v: serde_json::Value = serde_json::from_str(&f.body_json(&f.snapshot())).unwrap();
578        assert_eq!(v["agent_lines"], serde_json::json!({"AgentY": "Fixing the login race"}), "Gone is not listed");
579        let rev = f.snapshot().rev;
580        assert!(f.observe_fleet(with_lines(&[("AgentY", "Writing the tests")])), "a new line alone is a change");
581        assert_eq!(f.snapshot().rev, rev + 1);
582        assert!(f.observe_fleet(with_lines(&[])));
583        let v: serde_json::Value = serde_json::from_str(&f.body_json(&f.snapshot())).unwrap();
584        assert!(v.get("agent_lines").is_none(), "no lines, no field: {v}");
585    }
586
587    #[test]
588    fn a_state_change_bumps_rev() {
589        let f = feed();
590        assert!(f.observe_fleet(with_status(&["AgentY"], &[("AgentY", status(AgentState::Idle, 1))])));
591        assert_eq!(f.snapshot().rev, 2);
592        assert!(
593            !f.observe_fleet(with_status(&["AgentY"], &[("AgentY", status(AgentState::Idle, 1))])),
594            "same state, same start"
595        );
596        assert!(f.observe_fleet(with_status(&["AgentY"], &[("AgentY", status(AgentState::Working, 9))])));
597        assert_eq!(f.snapshot().rev, 3);
598        assert!(f.observe_fleet(with_status(&["AgentY"], &[])), "state no longer known");
599        assert_eq!(f.snapshot().rev, 4);
600        assert!(f.snapshot().agent_status.is_empty());
601    }
602
603    #[test]
604    fn the_install_id_is_reported_when_known() {
605        let f = FleetFeed::new("narko".into(), "stable".into(), "0.59.7".into())
606            .with_install_id(Some("mw3am46w5weex4a4fqrc3avnua".into()));
607        let body = f.body_json(&f.snapshot());
608        assert!(keys_of(&body).contains(&"install_id".to_string()));
609        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
610        assert_eq!(v["install_id"], "mw3am46w5weex4a4fqrc3avnua");
611    }
612
613    #[test]
614    fn a_kind_or_channel_count_change_bumps_rev_like_a_name_change() {
615        let f = feed();
616        assert!(f.observe_fleet(observation(&[("AgentX", Some("host"))], 1)));
617        assert_eq!(f.snapshot().rev, 2);
618        assert!(!f.observe_fleet(observation(&[("AgentX", Some("host"))], 1)), "nothing changed");
619        assert!(f.observe_fleet(observation(&[("AgentX", Some("container"))], 1)), "kind changed");
620        assert_eq!(f.snapshot().rev, 3);
621        assert!(f.observe_fleet(observation(&[("AgentX", Some("container"))], 3)), "a channel started");
622        assert_eq!(f.snapshot().rev, 4);
623        assert!(f.observe_fleet(observation(&[("AgentX", None)], 3)), "kind no longer known");
624        assert_eq!(f.snapshot().rev, 5);
625        assert!(f.snapshot().agent_kinds.is_empty());
626    }
627
628    #[test]
629    fn the_viewer_port_is_reported_while_up_and_its_change_bumps_rev() {
630        let port = Arc::new(AtomicU16::new(0));
631        let f = FleetFeed::new("h".into(), "c".into(), "v".into()).with_viewer_port(port.clone());
632        assert!(f.observe_fleet(observation(&[("AgentX", None)], 1)));
633        assert!(!keys_of(&f.body_json(&f.snapshot())).contains(&"viewer_port".to_string()));
634        port.store(29702, Ordering::Relaxed);
635        assert!(f.observe_fleet(observation(&[("AgentX", None)], 1)), "the port came up");
636        let v: serde_json::Value = serde_json::from_str(&f.body_json(&f.snapshot())).unwrap();
637        assert_eq!(v["viewer_port"], 29702);
638        assert!(!f.observe_fleet(observation(&[("AgentX", None)], 1)));
639        port.store(0, Ordering::Relaxed);
640        assert!(f.observe_fleet(observation(&[("AgentX", None)], 1)), "and went down");
641        assert_eq!(f.snapshot().viewer_port, None);
642    }
643
644    #[test]
645    fn kinds_follow_the_name_that_survives_deduplication_and_counts_are_clamped() {
646        let f = feed();
647        f.observe_fleet(observation(
648            &[("agenty", Some("container")), ("AgentY", Some("host")), ("Clamk", None)],
649            500,
650        ));
651        let snap = f.snapshot();
652        assert_eq!(*snap.agents, names(&["AgentY", "Clamk"]));
653        assert_eq!(
654            *snap.agent_kinds,
655            BTreeMap::from([("AgentY".to_string(), "host")]),
656            "the unknown kind is left out, not guessed"
657        );
658        assert_eq!(snap.channels_running, 99);
659        f.observe_fleet(observation(&[], 0));
660        assert_eq!(f.snapshot().channels_running, 1);
661    }
662
663    #[tokio::test]
664    async fn a_stream_starts_with_retry_then_the_current_state() {
665        let f = feed();
666        f.observe(names(&["AgentY"]));
667        let mut s = Box::pin(f.open_stream(None).unwrap());
668        assert_eq!(next_chunk(&mut s).await, "retry: 3000\n\n");
669        let event = next_chunk(&mut s).await;
670        let head = format!("event: fleet\nid: {}:2\ndata: ", f.epoch());
671        let data = event
672            .strip_prefix(&head)
673            .and_then(|rest| rest.strip_suffix("\n\n"))
674            .unwrap_or_else(|| panic!("{event}"));
675        let mut got: serde_json::Value = serde_json::from_str(data).unwrap();
676        let now_ms = got["now_ms"].as_u64().expect("now_ms");
677        got["now_ms"] = serde_json::json!(0);
678        let want: serde_json::Value = serde_json::from_str(&f.body_json_at(&f.snapshot(), 0)).unwrap();
679        assert_eq!(got, want);
680        assert!(now_ms > 0);
681    }
682
683    #[tokio::test]
684    async fn a_client_already_up_to_date_gets_no_initial_event_but_does_get_changes() {
685        let f = Arc::new(
686            FleetFeed::new("narko".into(), "stable".into(), "0.59.7".into())
687                .with_heartbeat(Duration::from_secs(3600)),
688        );
689        f.observe(names(&["AgentY"]));
690        let current = f.event_id(&f.snapshot());
691        let mut s = Box::pin(f.open_stream(Some(&current)).unwrap());
692        assert_eq!(next_chunk(&mut s).await, "retry: 3000\n\n");
693        // Nothing else is ready yet.
694        assert!(tokio::time::timeout(Duration::from_millis(200), s.next())
695            .await
696            .is_err());
697
698        f.observe(names(&["AgentY", "Clamk"]));
699        let event = next_chunk(&mut s).await;
700        assert!(
701            event.starts_with(&format!("event: fleet\nid: {}:3\n", f.epoch())),
702            "{event}"
703        );
704        assert!(event.contains(r#""agents":["AgentY","Clamk"]"#), "{event}");
705    }
706
707    #[tokio::test]
708    async fn a_stale_or_foreign_last_event_id_gets_the_snapshot() {
709        let f = feed();
710        for last in ["0000000000000000:1", "garbage", ""] {
711            let mut s = Box::pin(f.open_stream(Some(last)).unwrap());
712            next_chunk(&mut s).await;
713            assert!(
714                next_chunk(&mut s).await.starts_with("event: fleet\n"),
715                "{last}"
716            );
717        }
718    }
719
720    #[tokio::test]
721    async fn an_idle_stream_sends_heartbeats() {
722        let f = Arc::new(
723            FleetFeed::new("h".into(), "c".into(), "v".into())
724                .with_heartbeat(Duration::from_millis(50)),
725        );
726        let mut s = Box::pin(f.open_stream(None).unwrap());
727        next_chunk(&mut s).await;
728        next_chunk(&mut s).await;
729        assert_eq!(next_chunk(&mut s).await, ": hb\n\n");
730    }
731
732    #[test]
733    fn streams_are_capped_and_a_dropped_stream_frees_its_slot() {
734        let f = feed();
735        let open: Vec<_> = (0..MAX_STREAMS)
736            .map(|_| f.open_stream(None).expect("under the cap"))
737            .collect();
738        assert_eq!(f.open_streams(), MAX_STREAMS);
739        assert!(f.open_stream(None).is_none(), "the 33rd is refused");
740        drop(open);
741        assert_eq!(f.open_streams(), 0);
742        assert!(f.open_stream(None).is_some());
743    }
744
745    #[tokio::test]
746    async fn change_detection_publishes_and_stops_on_cancel() {
747        let f = feed();
748        let source = Arc::new(parking_lot::Mutex::new(names(&["a"])));
749        let token = tokio_util::sync::CancellationToken::new();
750        let read = source.clone();
751        f.clone().spawn_change_detection(
752            move || FleetObservation {
753                agents: read.lock().iter().map(|n| (n.clone(), None)).collect(),
754                channels_running: 1,
755                ..Default::default()
756            },
757            token.clone(),
758        );
759        assert_eq!(
760            *f.snapshot().agents,
761            names(&["a"]),
762            "primed before the first tick"
763        );
764
765        let mut rx = f.tx.subscribe();
766        *source.lock() = names(&["a", "b"]);
767        tokio::time::timeout(Duration::from_secs(5), rx.changed())
768            .await
769            .unwrap()
770            .unwrap();
771        assert_eq!(*f.snapshot().agents, names(&["a", "b"]));
772
773        token.cancel();
774        tokio::time::sleep(CHANGE_POLL_INTERVAL + Duration::from_millis(200)).await;
775        let rev = f.snapshot().rev;
776        *source.lock() = names(&["c"]);
777        tokio::time::sleep(CHANGE_POLL_INTERVAL * 2).await;
778        assert_eq!(f.snapshot().rev, rev, "no updates after cancel");
779    }
780}
agentmux_srv/backend/fleet_feed.rs

agentmux_srv/backend/
fleet_feed.rs
