majordomus_cli/mesh/
rendezvous.rs

1//! The rendezvous provider: discovery across segments multicast cannot cross. Every
2//! Majordomus server is a rendezvous — the `mesh.register` capability accepts a signed
3//! envelope and answers with the envelopes it holds — so a "mothership" is any node the
4//! declaration points at, not a different program and not a canonical database. The
5//! local registry stays the only truth; a rendezvous is one more source of observations,
6//! its failure stretches the retry and stops nothing else, and its answers are signed
7//! end-to-end: a poisoned rendezvous can withhold nodes, it cannot invent one.
8//!
9//! ```
10//! use majordomus_cli::mesh::rendezvous::RendezvousProvider;
11//! use majordomus_cli::mesh::provider::{MeshProvider, MeshProviderState};
12//!
13//! // Before the manager starts it, a provider is stopped; with no endpoints declared
14//! // there is nothing to register with, and starting it says so.
15//! let provider = RendezvousProvider::new(vec![], 60);
16//! assert_eq!(provider.id(), "rendezvous");
17//! assert_eq!(provider.status().state, MeshProviderState::Stopped);
18//! ```
19
20use std::sync::atomic::Ordering;
21use std::sync::{Arc, Mutex};
22use std::time::Duration;
23
24use super::protocol::Envelope;
25use super::provider::{
26    jitter_ms, Counters, MeshProvider, MeshProviderState, Observation, ProviderContext,
27    ProviderStatus,
28};
29use super::registry::MeshSource;
30use super::MeshError;
31
32/// The HTTP route `mesh.register` is served on. Named once, here: the capability
33/// declaration and this client read the same word.
34pub const REGISTER_PATH: &str = "/api/v1/mesh/register";
35
36/// One request's bound.
37const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
38
39/// The most interval multiples backoff stretches to after consecutive failures.
40const MAX_BACKOFF_FACTOR: u32 = 8;
41
42/// A stop check happens at least this often while waiting.
43const SLICE: Duration = Duration::from_millis(500);
44
45/// What a registration answers: whether the envelope was accepted, and the candidates
46/// the rendezvous holds. Defined here beside the client; the capability re-exports it.
47#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
48pub struct RegisterAnswer {
49    /// Whether the presented envelope was accepted into the local registry.
50    pub accepted: bool,
51    /// Why not, when it was not.
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub refusal: Option<String>,
54    /// The signed envelopes of the nodes this rendezvous currently holds, the answering
55    /// node's own included. Each verifies on its own signature; the rendezvous vouches
56    /// for nothing.
57    pub candidates: Vec<Envelope>,
58}
59
60/// The rendezvous provider: the declared endpoints, one registration thread each,
61/// and the counters their status reports. Constructed by the manager from the declaration.
62pub struct RendezvousProvider {
63    endpoints: Vec<String>,
64    interval: Duration,
65    counters: Arc<Counters>,
66    state: Arc<Mutex<(MeshProviderState, Option<String>)>>,
67}
68
69impl RendezvousProvider {
70    /// A provider over its endpoints; nothing is contacted until the manager calls
71    /// `start`, and each endpoint then gets its own bounded, backed-off loop.
72    pub fn new(endpoints: Vec<String>, interval_seconds: u64) -> Self {
73        RendezvousProvider {
74            endpoints,
75            interval: Duration::from_secs(interval_seconds),
76            counters: Arc::new(Counters::default()),
77            state: Arc::new(Mutex::new((MeshProviderState::Stopped, None))),
78        }
79    }
80}
81
82impl MeshProvider for RendezvousProvider {
83    fn id(&self) -> &'static str {
84        "rendezvous"
85    }
86
87    fn start(&mut self, ctx: &ProviderContext) -> Result<(), MeshError> {
88        if self.endpoints.is_empty() {
89            let reason = "no endpoints declared".to_string();
90            *self.state.lock().expect("rendezvous state") =
91                (MeshProviderState::Stopped, Some(reason.clone()));
92            return Err(MeshError::Provider(reason));
93        }
94        *self.state.lock().expect("rendezvous state") = (
95            MeshProviderState::Running,
96            Some(format!("{} endpoint(s)", self.endpoints.len())),
97        );
98        // One thread per endpoint: endpoints are independent sources, and one that
99        // hangs must not delay another. Each is bounded by REQUEST_TIMEOUT anyway.
100        for endpoint in self.endpoints.clone() {
101            let tx = ctx.tx.clone();
102            let stop = Arc::clone(&ctx.stop);
103            let beacon = Arc::clone(&ctx.beacon);
104            let counters = Arc::clone(&self.counters);
105            let interval = self.interval;
106            let _ = std::thread::Builder::new()
107                .name("majordomus-mesh-rendezvous".into())
108                .spawn(move || {
109                    let mut failures: u32 = 0;
110                    while !stop.load(Ordering::SeqCst) {
111                        match register_once(&endpoint, &beacon.next_envelope()) {
112                            Ok(answer) => {
113                                failures = 0;
114                                counters.sent.fetch_add(1, Ordering::Relaxed);
115                                for candidate in answer.candidates {
116                                    if let Ok(bytes) = serde_json::to_vec(&candidate) {
117                                        counters.received.fetch_add(1, Ordering::Relaxed);
118                                        let _ = tx.send(Observation {
119                                            source: MeshSource::Rendezvous,
120                                            path: endpoint.clone(),
121                                            bytes,
122                                        });
123                                    }
124                                }
125                            }
126                            Err(_) => {
127                                failures = (failures + 1).min(MAX_BACKOFF_FACTOR.ilog2());
128                            }
129                        }
130                        // interval × 2^failures, capped at ×MAX_BACKOFF_FACTOR: an
131                        // unreachable rendezvous is asked less and less, never hammered,
132                        // never a tight loop.
133                        let total =
134                            interval * (1u32 << failures) + Duration::from_millis(jitter_ms(2000));
135                        let mut slept = Duration::ZERO;
136                        while slept < total && !stop.load(Ordering::SeqCst) {
137                            let slice = SLICE.min(total - slept);
138                            std::thread::sleep(slice);
139                            slept += slice;
140                        }
141                    }
142                });
143        }
144        Ok(())
145    }
146
147    fn status(&self) -> ProviderStatus {
148        let (state, detail) = self.state.lock().expect("rendezvous state").clone();
149        ProviderStatus {
150            id: self.id().into(),
151            state,
152            detail,
153            sent: self.counters.sent.load(Ordering::Relaxed),
154            received: self.counters.received.load(Ordering::Relaxed),
155        }
156    }
157}
158
159/// One registration round-trip. Plain HTTP through the same minimal client the MCP
160/// bridge uses; any non-200, timeout or unparsable body is one failure.
161fn register_once(endpoint: &str, envelope: &Envelope) -> Result<RegisterAnswer, MeshError> {
162    let body = serde_json::json!({ "envelope": envelope }).to_string();
163    let reply = crate::mcp::bridge::request(
164        endpoint,
165        "POST",
166        REGISTER_PATH,
167        &[("Content-Type", "application/json")],
168        Some(&body),
169        REQUEST_TIMEOUT,
170    )
171    .map_err(|e| MeshError::Provider(format!("{endpoint}: {e}")))?;
172    if reply.status != 200 {
173        return Err(MeshError::Provider(format!(
174            "{endpoint}: status {}",
175            reply.status
176        )));
177    }
178    serde_json::from_str(&reply.body)
179        .map_err(|e| MeshError::Provider(format!("{endpoint}: not a register answer: {e}")))
180}
181
182#[cfg(test)]
183mod tests {
184    use super::*;
185
186    #[test]
187    fn no_endpoints_means_a_stopped_provider_with_the_reason() {
188        let mut provider = RendezvousProvider::new(vec![], 60);
189        let (tx, _rx) = crate::mesh::provider::ObservationSender::bounded(64);
190        let ctx = crate::mesh::provider::ProviderContext {
191            tx,
192            stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)),
193            beacon: std::sync::Arc::new(crate::mesh::provider::Beacon::new(
194                std::sync::Arc::new(crate::mesh::identity::NodeIdentity::ephemeral().unwrap()),
195                vec![],
196                vec![],
197                vec![],
198                "test",
199            )),
200        };
201        assert!(provider.start(&ctx).is_err());
202        let status = provider.status();
203        assert_eq!(status.state, MeshProviderState::Stopped);
204        assert_eq!(status.detail.as_deref(), Some("no endpoints declared"));
205    }
206
207    #[test]
208    fn an_unreachable_endpoint_is_one_failure_not_a_panic() {
209        let identity = crate::mesh::identity::NodeIdentity::ephemeral().unwrap();
210        let envelope = crate::mesh::protocol::advertise(&identity, 1, &[], &[], &[], "test");
211        // A port nothing listens on: connection refused, reported as the endpoint's error.
212        let error = register_once("http://127.0.0.1:9", &envelope)
213            .unwrap_err()
214            .to_string();
215        assert!(error.contains("127.0.0.1:9"), "{error}");
216    }
217
218    #[test]
219    fn a_reachable_endpoint_yields_registrations_and_candidate_observations() {
220        use std::io::{Read, Write};
221        // A canned rendezvous: one thread, plain HTTP/1.1, answering every POST with
222        // an accepted registration carrying one signed candidate.
223        let candidate_identity = crate::mesh::identity::NodeIdentity::ephemeral().unwrap();
224        let candidate = crate::mesh::protocol::advertise(
225            &candidate_identity,
226            1,
227            &["127.0.0.1:3".into()],
228            &[],
229            &[],
230            "canned",
231        );
232        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
233        let endpoint = format!("http://{}", listener.local_addr().unwrap());
234        let body = serde_json::to_string(&RegisterAnswer {
235            accepted: true,
236            refusal: None,
237            candidates: vec![candidate.clone()],
238        })
239        .unwrap();
240        std::thread::spawn(move || {
241            for stream in listener.incoming() {
242                let Ok(mut stream) = stream else { return };
243                let mut buffer = [0u8; 4096];
244                let _ = stream.read(&mut buffer);
245                let _ = stream.write_all(
246                    format!(
247                        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
248                        body.len(),
249                        body
250                    )
251                    .as_bytes(),
252                );
253            }
254        });
255
256        let mut provider = RendezvousProvider::new(vec![endpoint.clone()], 1);
257        let (tx, rx) = crate::mesh::provider::ObservationSender::bounded(64);
258        let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
259        let ctx = crate::mesh::provider::ProviderContext {
260            tx,
261            stop: std::sync::Arc::clone(&stop),
262            beacon: std::sync::Arc::new(crate::mesh::provider::Beacon::new(
263                std::sync::Arc::new(crate::mesh::identity::NodeIdentity::ephemeral().unwrap()),
264                vec![],
265                vec![],
266                vec![],
267                "test",
268            )),
269        };
270        provider.start(&ctx).unwrap();
271        let heard = rx
272            .recv_timeout(Duration::from_secs(10))
273            .expect("a candidate observation arrives");
274        assert_eq!(heard.source, MeshSource::Rendezvous);
275        assert_eq!(heard.path, endpoint);
276        let parsed = crate::mesh::protocol::parse(&heard.bytes).expect("the candidate verifies");
277        assert_eq!(
278            parsed.adv.node_id(),
279            Some(candidate_identity.public.node_id.clone())
280        );
281        assert!(provider.status().sent >= 1);
282        stop.store(true, std::sync::atomic::Ordering::SeqCst);
283    }
284
285    #[test]
286    fn a_register_answer_round_trips_as_json() {
287        let answer = RegisterAnswer {
288            accepted: true,
289            refusal: None,
290            candidates: vec![],
291        };
292        let text = serde_json::to_string(&answer).unwrap();
293        let back: RegisterAnswer = serde_json::from_str(&text).unwrap();
294        assert!(back.accepted && back.candidates.is_empty());
295    }
296}
majordomus_cli/mesh/rendezvous.rs

majordomus_cli/mesh/
rendezvous.rs
