agentmux_srv/backend/
fleet_source.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! What this install says about itself beyond agent names, on the LAN fleet
5//! feed, the UDP and mDNS records and the cloud presence record
6//! (agentmux-mobile's SPEC_FLEET_HOST_TAGS_AND_CLOUD_HOSTS_2026_10_06 §5–6):
7//! each agent's kind, how many channels this machine is running, and this
8//! install's id. Also each agent's state and since when
9//! (`backend::agent_state`, agentmux-mobile's
10//! SPEC_AGENT_STATUS_AND_LIVE_PANE_FEED_2026_10_07 §3).
11//!
12//! The fleet feed's change detection calls [`FleetSource::observe`] once a
13//! second. The LAN replies, which have no fleet feed in hand, read the
14//! channel count it last saw ([`latest_channels_running`]) and
15//! [`install_id`].
16
17use std::collections::{BTreeMap, HashMap};
18use std::sync::atomic::{AtomicU32, Ordering};
19use std::sync::{Arc, OnceLock};
20use std::time::{Duration, Instant};
21
22use crate::backend::agent_state::{AgentState, SinceTracker};
23use crate::backend::fleet_feed::{AgentKind, FleetObservation};
24use crate::backend::reactive::registry::AgentEntry;
25use crate::backend::storage::store::Store;
26
27/// The upper bound of `channels_running` on every wire.
28pub const MAX_CHANNELS_RUNNING: u32 = 99;
29
30/// How long a channel count is reused before the shared registry is read
31/// again. The registry's own heartbeat is 20 s; reading its directory every
32/// second would buy nothing.
33const CHANNELS_REREAD: Duration = Duration::from_secs(5);
34
35/// The count [`FleetSource::observe`] last saw; 0 before the first look.
36static LATEST_CHANNELS_RUNNING: AtomicU32 = AtomicU32::new(0);
37
38static INSTALL_ID: OnceLock<String> = OnceLock::new();
39
40/// This install's WAN instance id (26 chars of lowercase base32, a hash of
41/// the instance public key; not a secret), or `None` when this process has
42/// no WAN identity store. The instance is minted once per channel and never
43/// changes, so the first answer is kept.
44pub fn install_id() -> Option<String> {
45    if let Some(id) = INSTALL_ID.get() {
46        return Some(id.clone());
47    }
48    let wan = crate::backend::storage::wan_identity::global()?;
49    let instance = wan
50        .instance_ensure(&crate::backend::reactive::registry::local_host_label())
51        .ok()?;
52    Some(INSTALL_ID.get_or_init(|| instance.instance_id).clone())
53}
54
55/// 1 for this channel plus every other channel of this machine with a live
56/// entry in the host-global shared registry (`swarm_remote::other_channels`,
57/// with its `HIDE_AFTER_MS`), clamped to 1..=[`MAX_CHANNELS_RUNNING`]. A
58/// channel with no agent registered has no entry and is not counted.
59pub fn count_channels_running(entries: &[AgentEntry], own_channel: &str, own_url: &str, now_ms: u64) -> u32 {
60    let others = crate::backend::swarm_remote::other_channels(entries, &Default::default(), own_channel, own_url, now_ms).len();
61    u32::try_from(others)
62        .unwrap_or(u32::MAX)
63        .saturating_add(1)
64        .clamp(1, MAX_CHANNELS_RUNNING)
65}
66
67/// The channel count the fleet feed last saw, for the UDP replies; `None`
68/// before its first look.
69pub fn latest_channels_running() -> Option<u32> {
70    match LATEST_CHANNELS_RUNNING.load(Ordering::Relaxed) {
71        0 => None,
72        n => Some(n),
73    }
74}
75
76/// Each agent's kind by name, from its block. An agent whose block can't be
77/// read is left out (a peer then shows no tag rather than a guess).
78pub fn kinds_by_name<'a>(
79    agents: impl IntoIterator<Item = (&'a str, &'a str)>,
80    kind_of_block: impl Fn(&str) -> Option<AgentKind>,
81) -> BTreeMap<String, AgentKind> {
82    agents
83        .into_iter()
84        .filter_map(|(name, block_id)| kind_of_block(block_id).map(|k| (name.to_string(), k)))
85        .collect()
86}
87
88type AgentList = Box<dyn Fn() -> Vec<(String, String)> + Send + Sync>;
89type KindOfBlock = Box<dyn Fn(&str) -> Option<AgentKind> + Send + Sync>;
90type StateOfBlock = Box<dyn Fn(&str) -> Option<AgentState> + Send + Sync>;
91type LineOfBlock = Box<dyn Fn(&str) -> Option<String> + Send + Sync>;
92type RegistryRead = Box<dyn Fn() -> Vec<AgentEntry> + Send + Sync>;
93
94/// The fleet feed's view of this instance.
95pub struct FleetSource {
96    /// `(agent name, block id)` of every reachable agent.
97    agents: AgentList,
98    kind_of_block: KindOfBlock,
99    /// The block's state now (`agent_state::current_state`); read on every
100    /// look, unlike the kind.
101    state_of_block: StateOfBlock,
102    /// The block's summary line now (`agent_line`), `None` while sharing is
103    /// off; read on every look.
104    line_of_block: LineOfBlock,
105    /// When each state began; a block no longer listed is forgotten.
106    since: &'static SinceTracker,
107    shared_registry: RegistryRead,
108    own_channel: String,
109    own_url: String,
110    /// Kinds already read, by block id. A block's `agentMode` is fixed when
111    /// the agent is created, so each block is read once while it is listed.
112    kinds: parking_lot::Mutex<HashMap<String, AgentKind>>,
113    channels: parking_lot::Mutex<Option<(Instant, u32)>>,
114}
115
116impl FleetSource {
117    /// The live source: the reactive registry, the object store, the
118    /// settings and the host-global shared registry.
119    pub fn live(
120        handler: &'static crate::backend::reactive::handler::ReactiveHandler,
121        mstore: Arc<Store>,
122        config: Arc<crate::backend::wconfig::ConfigState>,
123        own_url: String,
124    ) -> Self {
125        let state_store = Arc::clone(&mstore);
126        let line_store = Arc::clone(&mstore);
127        Self::new(
128            Box::new(move || {
129                handler
130                    .list_agents()
131                    .into_iter()
132                    .map(|a| (a.agent_id, a.block_id))
133                    .collect()
134            }),
135            Box::new(move |block_id| crate::backend::operator_config_seed::agent_kind_of_block(&mstore, block_id)),
136            Box::new(move |block_id| crate::backend::agent_state::current_state(&state_store, block_id)),
137            Box::new(move |block_id| {
138                crate::backend::agent_line::sharing_on(&config.get_full_config().settings)
139                    .then(|| crate::backend::agent_line::agent_line_of_block(&line_store, block_id))
140                    .flatten()
141            }),
142            crate::backend::agent_state::global_tracker(),
143            Box::new(|| {
144                crate::registry::resolve_shared_reactive_dir()
145                    .map(|dir| crate::backend::reactive::registry::list_all_shared(&dir))
146                    .unwrap_or_default()
147            }),
148            crate::backend::reactive::registry::local_channel_id(),
149            own_url,
150        )
151    }
152
153    fn new(
154        agents: AgentList,
155        kind_of_block: KindOfBlock,
156        state_of_block: StateOfBlock,
157        line_of_block: LineOfBlock,
158        since: &'static SinceTracker,
159        shared_registry: RegistryRead,
160        own_channel: String,
161        own_url: String,
162    ) -> Self {
163        Self {
164            agents,
165            kind_of_block,
166            state_of_block,
167            line_of_block,
168            since,
169            shared_registry,
170            own_channel,
171            own_url,
172            kinds: parking_lot::Mutex::new(HashMap::new()),
173            channels: parking_lot::Mutex::new(None),
174        }
175    }
176
177    /// One look. Blocking: reads the store and the registry directory.
178    pub fn observe(&self) -> FleetObservation {
179        let listed = (self.agents)();
180        self.since.retain(|block_id| listed.iter().any(|(_, b)| b == block_id));
181        let now_ms = agentmux_common::time::now_ms_u64();
182        let agent_status = listed
183            .iter()
184            .filter_map(|(name, block_id)| {
185                let status = self.since.record(block_id, (self.state_of_block)(block_id), now_ms)?;
186                Some((name.clone(), status))
187            })
188            .collect();
189        let agent_lines = listed
190            .iter()
191            .filter_map(|(name, block_id)| Some((name.clone(), (self.line_of_block)(block_id)?)))
192            .collect();
193        let agents = {
194            let mut cache = self.kinds.lock();
195            cache.retain(|block_id, _| listed.iter().any(|(_, b)| b == block_id));
196            listed
197                .into_iter()
198                .map(|(name, block_id)| {
199                    let kind = match cache.get(&block_id) {
200                        Some(kind) => Some(*kind),
201                        None => {
202                            let kind = (self.kind_of_block)(&block_id);
203                            if let Some(kind) = kind {
204                                cache.insert(block_id, kind);
205                            }
206                            kind
207                        }
208                    };
209                    (name, kind)
210                })
211                .collect()
212        };
213        FleetObservation {
214            agents,
215            channels_running: self.channels_running(),
216            agent_status,
217            agent_lines,
218        }
219    }
220
221    fn channels_running(&self) -> u32 {
222        let mut last = self.channels.lock();
223        if let Some((at, n)) = *last {
224            if at.elapsed() < CHANNELS_REREAD {
225                return n;
226            }
227        }
228        let n = count_channels_running(
229            &(self.shared_registry)(),
230            &self.own_channel,
231            &self.own_url,
232            agentmux_common::time::now_ms_u64(),
233        );
234        *last = Some((Instant::now(), n));
235        LATEST_CHANNELS_RUNNING.store(n, Ordering::Relaxed);
236        n
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243    use std::sync::atomic::AtomicUsize;
244
245    const NOW: u64 = 1_000_000_000;
246
247    fn entry(name: &str, channel: &str, age_ms: u64) -> AgentEntry {
248        AgentEntry {
249            agent_id: name.to_string(),
250            local_url: format!("http://127.0.0.1/{channel}"),
251            block_id: format!("block-{name}"),
252            pid: 1,
253            updated_at: NOW - age_ms,
254            auth_key: String::new(),
255            channel: channel.to_string(),
256            registration_nonce: 0,
257            jekt_public_key: String::new(),
258            uid: String::new(),
259            uid_public_key: String::new(),
260        }
261    }
262
263    #[test]
264    fn channels_running_counts_this_channel_and_live_others() {
265        assert_eq!(count_channels_running(&[], "stable", "", NOW), 1);
266        let entries = [
267            entry("a", "stable", 0),
268            entry("b", "dev", 0),
269            entry("c", "dev", 0),
270            entry("d", "local-main", 1_000),
271            entry("e", "crashed", crate::backend::swarm_remote::HIDE_AFTER_MS + 1),
272        ];
273        assert_eq!(
274            count_channels_running(&entries, "stable", "", NOW),
275            3,
276            "this one, dev and local-main; not the hidden one"
277        );
278        let many: Vec<_> = (0..150).map(|i| entry("x", &format!("ch-{i}"), 0)).collect();
279        assert_eq!(count_channels_running(&many, "stable", "", NOW), MAX_CHANNELS_RUNNING);
280    }
281
282    #[test]
283    fn kinds_leave_out_unreadable_blocks() {
284        let kinds = kinds_by_name([("AgentX", "b1"), ("Camper", "b2"), ("Ghost", "b3")], |b| match b {
285            "b1" => Some("container"),
286            "b2" => Some("host"),
287            _ => None,
288        });
289        assert_eq!(
290            kinds,
291            BTreeMap::from([("AgentX".to_string(), "container"), ("Camper".to_string(), "host")])
292        );
293    }
294
295    #[test]
296    fn a_block_is_read_once_while_listed_and_an_unreadable_one_is_retried() {
297        let reads = Arc::new(AtomicUsize::new(0));
298        let counted = reads.clone();
299        let ready = Arc::new(std::sync::atomic::AtomicBool::new(false));
300        let ready_read = ready.clone();
301        let source = FleetSource::new(
302            Box::new(|| vec![("AgentX".into(), "b1".into()), ("Late".into(), "b2".into())]),
303            Box::new(move |b| {
304                counted.fetch_add(1, Ordering::SeqCst);
305                match b {
306                    "b1" => Some("container"),
307                    _ if ready_read.load(Ordering::SeqCst) => Some("host"),
308                    _ => None,
309                }
310            }),
311            Box::new(|_| None),
312            Box::new(|_| None),
313            tracker(),
314            Box::new(Vec::new),
315            "stable".into(),
316            String::new(),
317        );
318        let first = source.observe();
319        assert_eq!(
320            first.agents,
321            vec![("AgentX".to_string(), Some("container")), ("Late".to_string(), None)]
322        );
323        assert_eq!(first.channels_running, 1);
324        assert_eq!(reads.load(Ordering::SeqCst), 2);
325        source.observe();
326        assert_eq!(reads.load(Ordering::SeqCst), 3, "b1 cached, b2 retried");
327        ready.store(true, Ordering::SeqCst);
328        assert_eq!(source.observe().agents[1], ("Late".to_string(), Some("host")));
329        source.observe();
330        assert_eq!(reads.load(Ordering::SeqCst), 4, "both cached now");
331    }
332
333    /// A tracker of the test's own, not the live one the names route shares.
334    fn tracker() -> &'static SinceTracker {
335        Box::leak(Box::new(SinceTracker::default()))
336    }
337
338    #[test]
339    fn states_are_read_every_look_and_keep_their_start_while_unchanged() {
340        let states = Arc::new(parking_lot::Mutex::new(HashMap::from([
341            ("b1".to_string(), AgentState::Working),
342            ("b2".to_string(), AgentState::Idle),
343        ])));
344        let read = states.clone();
345        let listed = Arc::new(parking_lot::Mutex::new(vec![
346            ("AgentX".to_string(), "b1".to_string()),
347            ("Camper".to_string(), "b2".to_string()),
348            ("Term".to_string(), "b3".to_string()),
349        ]));
350        let list = listed.clone();
351        let since = tracker();
352        let source = FleetSource::new(
353            Box::new(move || list.lock().clone()),
354            Box::new(|_| Some("host")),
355            Box::new(move |b| read.lock().get(b).copied()),
356            Box::new(|_| None),
357            since,
358            Box::new(Vec::new),
359            "stable".into(),
360            String::new(),
361        );
362        let first = source.observe().agent_status;
363        assert_eq!(
364            first.keys().collect::<Vec<_>>(),
365            ["AgentX", "Camper"],
366            "the agent with no state is left out"
367        );
368        assert_eq!(first["AgentX"].state, AgentState::Working);
369
370        std::thread::sleep(Duration::from_millis(5));
371        let second = source.observe().agent_status;
372        assert_eq!(second, first, "nothing changed, the starts held");
373
374        states.lock().insert("b1".into(), AgentState::Idle);
375        let third = source.observe().agent_status;
376        assert_eq!(third["AgentX"].state, AgentState::Idle);
377        assert!(third["AgentX"].since_ms > first["AgentX"].since_ms, "a change starts again");
378        assert_eq!(third["Camper"], first["Camper"]);
379
380        listed.lock().remove(0);
381        source.observe();
382        assert_eq!(since.len(), 1, "an agent no longer listed is forgotten");
383    }
384}
agentmux_srv/backend/fleet_source.rs

agentmux_srv/backend/
fleet_source.rs
