umsh/
ulcp_mesh.rs

1//! ULCP carried over the Node Management binding.
2//!
3//! [`UlcpDevice`](crate::ulcp::UlcpDevice) speaks to whatever satisfies
4//! [`FrameLink`]: send one frame, receive one frame. That is also the
5//! whole of what an administrative exchange does, so a link made of the
6//! two lets the ordinary host surface — properties, capabilities, the
7//! synchronization procedure, every category command built on them —
8//! reach a device across the mesh with nothing about it rewritten for
9//! the occasion.
10//!
11//! The adapter is split in half. [`MeshFrameLink`] is what the device
12//! handle holds; [`MeshEndpoint`] is what the driver that owns the radio
13//! holds. Between them they reconcile the two differences between a wire
14//! and the binding:
15//!
16//! - **Transaction identifiers.** A wire correlates a response to its
17//!   request by TID. The binding requires TID 0 on every frame and
18//!   correlates by envelope token instead, so requests go out with their
19//!   TID cleared and replies come back wearing the one their request
20//!   carried.
21//! - **Reset-class commands.** `CMD_RST`, `CMD_RESTORE` and
22//!   `CMD_FACTORY_RESET` are answered over the binding by no payload at
23//!   all — delivery is the acknowledgment. A local device announces its
24//!   new status instead, and callers wait for that announcement, so this
25//!   synthesizes the announcement the device would have sent.
26//!
27//!   One consequence is worth knowing about. The exchange engine decides
28//!   a command is reset-class by reading the *request*, and settles it as
29//!   soon as the acknowledgment lands — before any reply is drained. A
30//!   `CMD_RESTORE` that the device refuses for want of a snapshot does
31//!   answer, with `STATUS_INVALID_STATE`, and that answer loses the race:
32//!   the caller is told the restore completed. Reading
33//!   `PROP_LAST_STATUS` in a later exchange is how the spec says to find
34//!   out what actually happened, and it remains the way to tell these two
35//!   apart over the mesh.
36//!
37//! Everything else the binding refuses — the host domain, session state,
38//! the private key — it refuses as `STATUS_PROP_NOT_FOUND`, which is an
39//! answer and needs no help from here.
40
41use tokio::sync::mpsc;
42
43use umsh_node_mgmt::admin::Failure;
44use umsh_ulcp::Status;
45use umsh_ulcp::frame::{self, Cmd, Frame, HEADER_FLG_PATTERN, HEADER_TID_MASK, TID_UNSOLICITED};
46
47use crate::ulcp::{FrameLink, UlcpError};
48
49/// Largest synthesized frame: a header, a command, and a status.
50const SYNTHETIC_BUF: usize = 8;
51
52/// Why an exchange produced no frame to hand back.
53///
54/// A fault is per-request, not terminal — the link stays usable for the
55/// next command — except [`MeshFault::Radio`], which ends the session
56/// because the radio underneath it is gone.
57#[derive(Clone, Debug)]
58pub enum MeshFault {
59    /// The exchange failed, already rendered as prose for whoever is
60    /// holding the tool.
61    Exchange(String),
62    /// The request is larger than one Node Management payload carries.
63    RequestTooLarge(usize),
64    /// The borrowed radio stopped answering; the session is over.
65    Radio(String),
66}
67
68impl From<MeshFault> for UlcpError {
69    fn from(fault: MeshFault) -> Self {
70        match fault {
71            // Not `Timeout`: the prose is the actionable part, and a
72            // bare timeout would throw it away.
73            MeshFault::Exchange(message) | MeshFault::Radio(message) => Self::Transport(message),
74            MeshFault::RequestTooLarge(len) => Self::FrameTooLarge(len),
75        }
76    }
77}
78
79/// One request the driver is to carry, and the bookkeeping needed to
80/// deliver its outcome.
81#[derive(Clone, Debug)]
82pub struct MeshRequest {
83    /// The request frame, its TID already cleared for the binding.
84    frame: Vec<u8>,
85    /// The TID the caller used, and the one its reply must wear.
86    tid: u8,
87    /// The command, when it is one this crate defines. `None` is not an
88    /// error — the binding carries whatever the caller encoded.
89    cmd: Option<Cmd>,
90}
91
92impl MeshRequest {
93    /// The frame to put on the air, with TID 0 as the binding requires.
94    pub fn frame(&self) -> &[u8] {
95        &self.frame
96    }
97
98    /// Whether the binding answers this command with no payload, so a
99    /// caller can tell an expected silence from a lost exchange.
100    pub fn is_reset_class(&self) -> bool {
101        matches!(
102            self.cmd,
103            Some(Cmd::Reset | Cmd::Restore | Cmd::FactoryReset)
104        )
105    }
106}
107
108/// How an exchange ended.
109#[derive(Clone, Copy, Debug)]
110pub enum DeliveredOutcome<'a> {
111    /// The device answered with this reply payload.
112    Replied(&'a [u8]),
113    /// The device answered nothing, its delivery acknowledged.
114    NoResponse,
115    /// The exchange failed.
116    Failed(Failure),
117}
118
119/// Build a mesh link and the endpoint that serves it.
120pub fn mesh_link() -> (MeshFrameLink, MeshEndpoint) {
121    let (request_tx, request_rx) = mpsc::unbounded_channel();
122    let (reply_tx, reply_rx) = mpsc::unbounded_channel();
123    (
124        MeshFrameLink {
125            requests: request_tx,
126            replies: reply_rx,
127        },
128        MeshEndpoint {
129            requests: request_rx,
130            replies: reply_tx,
131        },
132    )
133}
134
135/// The device handle's half: a [`FrameLink`] whose wire is an
136/// administrative exchange.
137pub struct MeshFrameLink {
138    requests: mpsc::UnboundedSender<Vec<u8>>,
139    replies: mpsc::UnboundedReceiver<Result<Vec<u8>, MeshFault>>,
140}
141
142impl FrameLink for MeshFrameLink {
143    async fn send_frame(&mut self, frame: &[u8]) -> Result<(), UlcpError> {
144        self.requests
145            .send(frame.to_vec())
146            .map_err(|_| UlcpError::Disconnected)
147    }
148
149    fn poll_recv_frame(
150        &mut self,
151        cx: &mut core::task::Context<'_>,
152    ) -> core::task::Poll<Result<Vec<u8>, UlcpError>> {
153        self.replies.poll_recv(cx).map(|received| match received {
154            Some(Ok(frame)) => Ok(frame),
155            Some(Err(fault)) => Err(fault.into()),
156            // The driver is gone, and with it the radio.
157            None => Err(UlcpError::Disconnected),
158        })
159    }
160}
161
162/// The driver's half: requests to carry, and outcomes to report.
163pub struct MeshEndpoint {
164    requests: mpsc::UnboundedReceiver<Vec<u8>>,
165    replies: mpsc::UnboundedSender<Result<Vec<u8>, MeshFault>>,
166}
167
168impl MeshEndpoint {
169    /// The next request to carry, or `None` once the link is dropped and
170    /// everything already queued has been handed over.
171    ///
172    /// A request too large for one payload is refused here rather than
173    /// reaching the exchange engine, so the caller sees the size it
174    /// asked for named in the error.
175    pub async fn next(&mut self) -> Option<MeshRequest> {
176        loop {
177            let mut frame = self.requests.recv().await?;
178            let Ok(parsed) = Frame::parse(&frame) else {
179                // A frame the grammar does not recognize cannot be
180                // classified, tokenized, or answered. Nothing built on
181                // this link produces one.
182                self.report(Err(MeshFault::Exchange(
183                    "the request is not a well-formed ULCP frame".into(),
184                )));
185                continue;
186            };
187            let tid = parsed.header.tid();
188            let cmd = parsed.command();
189            if frame.len() > umsh_node_mgmt::REQUEST_MAX {
190                let len = frame.len();
191                self.report(Err(MeshFault::RequestTooLarge(len)));
192                continue;
193            }
194            // The binding requires TID 0 and correlates by token; the
195            // reply wears the caller's TID again on the way back.
196            frame[0] = HEADER_FLG_PATTERN;
197            return Some(MeshRequest { frame, tid, cmd });
198        }
199    }
200
201    /// Hand one exchange's outcome back to the device handle.
202    pub fn deliver(&mut self, request: &MeshRequest, outcome: DeliveredOutcome<'_>) {
203        match outcome {
204            DeliveredOutcome::Replied(reply) => {
205                let mut reply = reply.to_vec();
206                if reply.is_empty() {
207                    self.report(Err(MeshFault::Exchange(
208                        "the device answered with an empty frame".into(),
209                    )));
210                    return;
211                }
212                reply[0] = HEADER_FLG_PATTERN | (request.tid & HEADER_TID_MASK);
213                self.report(Ok(reply));
214            }
215            DeliveredOutcome::NoResponse => self.synthesize(request),
216            DeliveredOutcome::Failed(failure) => {
217                self.report(Err(MeshFault::Exchange(describe(failure))))
218            }
219        }
220    }
221
222    /// Report that this request could not be carried, without ending the
223    /// session.
224    ///
225    /// For the failures that belong to one command rather than to the
226    /// link — a request that ran out of patience, an engine that would
227    /// not take it — where the next command may well succeed.
228    pub fn refuse(&mut self, message: String) {
229        self.report(Err(MeshFault::Exchange(message)));
230    }
231
232    /// Report a terminal failure and close the link. The device handle
233    /// sees this fault once, then `Disconnected`.
234    pub fn fail(self, fault: MeshFault) {
235        let _ = self.replies.send(Err(fault));
236    }
237
238    /// Stand in for the announcement a local device would have made.
239    ///
240    /// Delivery was the acknowledgment; the caller is waiting on the
241    /// status that a wire-attached device would have volunteered. It is
242    /// sent unsolicited (TID 0), which is what the waiting caller
243    /// watches for.
244    fn synthesize(&mut self, request: &MeshRequest) {
245        let status = match request.cmd {
246            Some(Cmd::Reset) => Status::RESET_SOFTWARE,
247            Some(Cmd::Restore) => Status::RESET_RESTORED,
248            // A factory reset is not waited on at all: the device wipes
249            // itself and reboots, and the link dropping is the report.
250            Some(Cmd::FactoryReset) => return,
251            // Unreachable in practice — the exchange engine only reports
252            // `NoResponse` for the three above — but silence here would
253            // hang the caller until its own timeout, which is a worse
254            // way to learn about a bug.
255            _ => {
256                self.report(Err(MeshFault::Exchange(
257                    "the device answered nothing where a reply was due".into(),
258                )));
259                return;
260            }
261        };
262        let mut buf = [0u8; SYNTHETIC_BUF];
263        match frame::last_status(&mut buf, TID_UNSOLICITED, status) {
264            Ok(len) => self.report(Ok(buf[..len].to_vec())),
265            Err(_) => self.report(Err(MeshFault::Exchange(
266                "could not report the device's completion".into(),
267            ))),
268        }
269    }
270
271    /// A closed channel means the caller gave up on this request; the
272    /// exchange still ran, and the next `next()` will end the loop.
273    fn report(&mut self, outcome: Result<Vec<u8>, MeshFault>) {
274        let _ = self.replies.send(outcome);
275    }
276}
277
278/// What a failed exchange means to somebody holding the tool.
279pub fn describe(failure: Failure) -> String {
280    match failure {
281        Failure::TimedOut => {
282            "no answer — the device may be out of range, or this host may not be one of its \
283             administrators"
284                .into()
285        }
286        Failure::CursorInvalid => {
287            "the device's state changed mid-read; run the command again".into()
288        }
289        Failure::TooLarge => "the answer is larger than this host reassembles".into(),
290        Failure::Malformed => "the device's answer could not be read".into(),
291        Failure::UnknownCriticalOption(number) => {
292            format!("the device's answer carries option {number}, which this host does not know")
293        }
294    }
295}
296
297#[cfg(test)]
298mod tests {
299    use super::*;
300    use umsh_ulcp::ids::prop;
301
302    /// Drive one request through the endpoint and collect what the link
303    /// is handed back.
304    async fn round_trip(
305        request: Vec<u8>,
306        answer: impl FnOnce(&MeshRequest) -> Option<Vec<u8>>,
307    ) -> (MeshFrameLink, Option<MeshRequest>) {
308        let (mut link, mut endpoint) = mesh_link();
309        link.send_frame(&request).await.unwrap();
310        let carried = endpoint.next().await;
311        if let Some(carried) = &carried {
312            match answer(carried) {
313                Some(reply) => endpoint.deliver(carried, DeliveredOutcome::Replied(&reply)),
314                None => endpoint.deliver(carried, DeliveredOutcome::NoResponse),
315            }
316        }
317        // Keep the endpoint alive past the reads the test makes.
318        core::mem::forget(endpoint);
319        (link, carried)
320    }
321
322    #[tokio::test]
323    async fn the_binding_sees_tid_zero_and_the_caller_sees_its_own() {
324        let mut buf = [0u8; 8];
325        let len = frame::prop_get(&mut buf, 5, prop::DEV_NAME).unwrap();
326
327        let (mut link, carried) = round_trip(buf[..len].to_vec(), |carried| {
328            // What goes on the air carries TID 0, as the binding requires.
329            assert_eq!(carried.frame()[0] & HEADER_TID_MASK, 0);
330            // The device answers with TID 0 too.
331            let mut reply = [0u8; 32];
332            let len =
333                frame::prop_is(&mut reply, TID_UNSOLICITED, prop::DEV_NAME, b"node\0").unwrap();
334            Some(reply[..len].to_vec())
335        })
336        .await;
337
338        assert!(!carried.unwrap().is_reset_class());
339        let reply = link.recv_frame().await.unwrap();
340        let parsed = Frame::parse(&reply).unwrap();
341        // ... and comes back wearing the TID the caller allocated, which
342        // is the only thing that files it as a response.
343        assert_eq!(parsed.header.tid(), 5);
344        assert_eq!(parsed.command(), Some(Cmd::PropIs));
345    }
346
347    #[tokio::test]
348    async fn a_reset_is_answered_by_the_announcement_a_wire_would_have_carried() {
349        let mut buf = [0u8; 8];
350        let len = frame::reset(&mut buf, TID_UNSOLICITED).unwrap();
351
352        let (mut link, carried) = round_trip(buf[..len].to_vec(), |_| None).await;
353
354        assert!(carried.unwrap().is_reset_class());
355        let reply = link.recv_frame().await.unwrap();
356        let parsed = Frame::parse(&reply).unwrap();
357        assert_eq!(parsed.header.tid(), TID_UNSOLICITED);
358        assert_eq!(parsed.command(), Some(Cmd::PropIs));
359        // The caller is waiting on a reset announcement; this is one.
360        assert_eq!(
361            umsh_ulcp::reply::status_of(&reply),
362            Some(Status::RESET_SOFTWARE)
363        );
364    }
365
366    #[tokio::test]
367    async fn a_restore_that_reset_announces_the_snapshot_it_came_up_on() {
368        let mut buf = [0u8; 8];
369        let len = frame::restore(&mut buf, 3).unwrap();
370
371        let (mut link, _) = round_trip(buf[..len].to_vec(), |_| None).await;
372
373        let reply = link.recv_frame().await.unwrap();
374        assert_eq!(
375            umsh_ulcp::reply::status_of(&reply),
376            Some(Status::RESET_RESTORED)
377        );
378    }
379
380    #[tokio::test]
381    async fn a_factory_reset_is_answered_by_nothing_at_all() {
382        let mut buf = [0u8; 8];
383        let len = frame::factory_reset(&mut buf, TID_UNSOLICITED).unwrap();
384        let (mut link, _) = round_trip(buf[..len].to_vec(), |_| None).await;
385
386        // The device wipes itself and reboots; there is nothing to wait
387        // for, and nothing is sent.
388        assert!(
389            tokio::time::timeout(core::time::Duration::from_millis(50), link.recv_frame())
390                .await
391                .is_err()
392        );
393    }
394
395    #[tokio::test]
396    async fn a_refused_command_costs_the_session_nothing() {
397        let (mut link, mut endpoint) = mesh_link();
398        let mut buf = [0u8; 8];
399        let len = frame::prop_get(&mut buf, 2, prop::DEV_NAME).unwrap();
400        link.send_frame(&buf[..len]).await.unwrap();
401        let carried = endpoint.next().await.unwrap();
402        assert_eq!(carried.frame()[0] & HEADER_TID_MASK, 0);
403        endpoint.refuse("gave up after 180 s".into());
404
405        match link.recv_frame().await {
406            Err(UlcpError::Transport(message)) => assert!(message.contains("gave up")),
407            other => panic!("expected a transport fault, got {other:?}"),
408        }
409        // One command ran out of patience; the session did not. A radio
410        // that has actually died is reported by `fail`, not by this.
411        link.send_frame(&buf[..len]).await.unwrap();
412        assert!(endpoint.next().await.is_some());
413    }
414
415    #[tokio::test]
416    async fn a_request_too_large_for_one_payload_names_its_size() {
417        let (mut link, mut endpoint) = mesh_link();
418        let mut buf = vec![0u8; umsh_node_mgmt::REQUEST_MAX + 64];
419        let oversize = vec![0xAAu8; umsh_node_mgmt::REQUEST_MAX];
420        let len = frame::prop_set(&mut buf, 1, prop::DEV_NAME, &oversize).unwrap();
421        link.send_frame(&buf[..len]).await.unwrap();
422
423        // The request never reaches the air.
424        assert!(
425            tokio::time::timeout(core::time::Duration::from_millis(50), endpoint.next())
426                .await
427                .is_err()
428        );
429        match link.recv_frame().await {
430            Err(UlcpError::FrameTooLarge(reported)) => assert_eq!(reported, len),
431            other => panic!("expected FrameTooLarge, got {other:?}"),
432        }
433    }
434
435    #[tokio::test]
436    async fn a_failed_exchange_keeps_its_prose_and_leaves_the_link_usable() {
437        let (mut link, mut endpoint) = mesh_link();
438        let mut buf = [0u8; 8];
439        let len = frame::prop_get(&mut buf, 1, prop::DEV_NAME).unwrap();
440        link.send_frame(&buf[..len]).await.unwrap();
441        let carried = endpoint.next().await.unwrap();
442        endpoint.deliver(&carried, DeliveredOutcome::Failed(Failure::TimedOut));
443
444        match link.recv_frame().await {
445            Err(UlcpError::Transport(message)) => assert!(message.contains("administrators")),
446            other => panic!("expected a transport fault, got {other:?}"),
447        }
448        // The fault was about the request, not the link: the next one
449        // goes out as usual.
450        link.send_frame(&buf[..len]).await.unwrap();
451        assert!(endpoint.next().await.is_some());
452    }
453
454    #[tokio::test]
455    async fn losing_the_driver_disconnects_the_handle() {
456        let (mut link, endpoint) = mesh_link();
457        drop(endpoint);
458        assert!(matches!(
459            link.recv_frame().await,
460            Err(UlcpError::Disconnected)
461        ));
462    }
463
464    #[tokio::test]
465    async fn a_terminal_radio_failure_is_reported_before_the_disconnect() {
466        let (mut link, endpoint) = mesh_link();
467        endpoint.fail(MeshFault::Radio("the radio stopped answering".into()));
468        match link.recv_frame().await {
469            Err(UlcpError::Transport(message)) => assert!(message.contains("stopped answering")),
470            other => panic!("expected a transport fault, got {other:?}"),
471        }
472        assert!(matches!(
473            link.recv_frame().await,
474            Err(UlcpError::Disconnected)
475        ));
476    }
477}
umsh/ulcp_mesh.rs

umsh/
ulcp_mesh.rs
