koan_core/remote/
acks.rs

1//! Commands a device answers once it has acted on them, so a sender knows a
2//! command arrived rather than only that it was queued.
3//!
4//! A command that wants an answer carries an id beside it on the wire
5//! (`Envelope`), never inside the `LinkCommand`: a server that predates this
6//! parses and re-serialises relayed commands, and would drop it, while a
7//! client that predates it reads the command and ignores the id. The target
8//! answers with the id and an `AckOutcome`.
9//!
10//! A sender that hears nothing tries the next way through with the same id,
11//! so a command can arrive twice. Each id is drawn at random, so none is
12//! reused across a sender's restarts and none can be guessed from another
13//! seen on the network, and a target remembers the ids it has taken for a
14//! minute: a repeat is answered with the first one's outcome and not acted on
15//! again. Only a command the target accepts from its source takes an id, so a
16//! stranger's refused command cannot reserve one.
17
18use std::collections::HashMap;
19use std::sync::atomic::{AtomicU64, Ordering};
20use std::sync::{Arc, LazyLock};
21use std::time::{Duration, Instant};
22
23use parking_lot::Mutex;
24use serde::{Deserialize, Serialize};
25
26use crate::remote::link::LinkCommand;
27
28/// How a command went, as the device it was sent to says.
29#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
30#[serde(tag = "result", rename_all = "camelCase")]
31pub enum AckOutcome {
32    /// Acted on.
33    Done,
34    /// Not acted on yet, but kept for the device: it is asleep, and takes it
35    /// when woken. Said by the server, for the device.
36    Queued,
37    /// Not allowed from where it came.
38    Refused { reason: String },
39    /// Taken, and it failed.
40    Failed { error: String },
41}
42
43/// A command as it travels, with the id that asks for an answer.
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct Envelope {
46    #[serde(flatten)]
47    pub command: LinkCommand,
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    pub ack: Option<u64>,
50}
51
52impl Envelope {
53    /// A command as a push carries it: the same JSON as over the link.
54    pub fn parse(json: &str) -> Result<Self, String> {
55        serde_json::from_str(json).map_err(|e| e.to_string())
56    }
57}
58
59impl From<LinkCommand> for Envelope {
60    fn from(command: LinkCommand) -> Self {
61        Self { command, ack: None }
62    }
63}
64
65/// How long a target remembers an id it has taken.
66const REMEMBERED: Duration = Duration::from_secs(60);
67
68// --- Sending ----------------------------------------------------------------
69
70/// A fresh id for a command that wants an answer: random, and never 0.
71pub fn next_id() -> u64 {
72    let mut bytes = [0u8; 8];
73    if getrandom::fill(&mut bytes).is_err() {
74        // No randomness to be had: unguessable no longer, but still unique
75        // within the run.
76        static FALLBACK: AtomicU64 = AtomicU64::new(1);
77        return FALLBACK.fetch_add(1, Ordering::Relaxed) | (1 << 63);
78    }
79    u64::from_le_bytes(bytes).max(1)
80}
81
82/// The device a command was sent to, and where its answer goes.
83type Waiter = (String, crossbeam_channel::Sender<AckOutcome>);
84
85/// Each waiting id, with its waiter.
86static WAITING: LazyLock<Mutex<HashMap<u64, Waiter>>> = LazyLock::new(Default::default);
87
88/// Wait for `device`'s answer to the command sent under `id`. Dropping the
89/// receiver stops waiting.
90pub fn expect(id: u64, device: &str) -> crossbeam_channel::Receiver<AckOutcome> {
91    let (tx, rx) = crossbeam_channel::bounded(1);
92    WAITING.lock().insert(id, (device.to_owned(), tx));
93    rx
94}
95
96/// `from` answered `id`, by whatever way. Only the device the command was
97/// sent to can answer it: nearby connections are not encrypted, so an id is
98/// seen by anyone on the network, and any device this one has dialled could
99/// otherwise report a command done that never arrived, or refused one that
100/// did.
101pub fn resolve(id: u64, from: &str, outcome: AckOutcome) {
102    let mut waiting = WAITING.lock();
103    match waiting.get(&id) {
104        Some((device, _)) if device == from => {}
105        Some((device, _)) => {
106            log::warn!("acks: {from} answered a command sent to {device}; ignored");
107            return;
108        }
109        None => return,
110    }
111    if let Some((_, tx)) = waiting.remove(&id) {
112        let _ = tx.try_send(outcome);
113    }
114}
115
116/// No longer waiting for `id`.
117pub fn forget(id: u64) {
118    WAITING.lock().remove(&id);
119}
120
121// --- Receiving --------------------------------------------------------------
122
123type Reply = Box<dyn FnOnce(AckOutcome) + Send>;
124
125enum Taken {
126    Running(Vec<Reply>),
127    Answered(AckOutcome),
128}
129
130static TAKEN: LazyLock<Mutex<HashMap<u64, (Instant, Taken)>>> = LazyLock::new(Default::default);
131
132/// A command taken under an id: finish it with its outcome, which answers the
133/// sender and any repeat that arrived meanwhile. Dropped unfinished, it
134/// answers that it failed.
135pub struct Pending {
136    id: u64,
137    finished: bool,
138}
139
140impl Pending {
141    pub fn finish(mut self, outcome: AckOutcome) {
142        self.finished = true;
143        answer(self.id, outcome);
144    }
145}
146
147impl Drop for Pending {
148    fn drop(&mut self) {
149        if !self.finished {
150            answer(
151                self.id,
152                AckOutcome::Failed {
153                    error: "it stopped before finishing".into(),
154                },
155            );
156        }
157    }
158}
159
160impl std::fmt::Debug for Pending {
161    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
162        write!(f, "Pending({})", self.id)
163    }
164}
165
166fn answer(id: u64, outcome: AckOutcome) {
167    let replies = {
168        let mut taken = TAKEN.lock();
169        match taken.insert(id, (Instant::now(), Taken::Answered(outcome.clone()))) {
170            Some((_, Taken::Running(replies))) => replies,
171            _ => Vec::new(),
172        }
173    };
174    for reply in replies {
175        reply(outcome.clone());
176    }
177}
178
179/// A command has arrived under `id`, to be answered through `reply`. The
180/// first time, it is to be acted on: `Some`, to finish once it has been. A
181/// repeat is not: `None`, and `reply` gets the first one's outcome, now if
182/// there is one, else when it is finished.
183pub fn accept(id: u64, reply: impl FnOnce(AckOutcome) + Send + 'static) -> Option<Pending> {
184    let mut taken = TAKEN.lock();
185    taken.retain(|_, (at, _)| at.elapsed() < REMEMBERED);
186    match taken.get_mut(&id) {
187        Some((_, Taken::Running(replies))) => {
188            replies.push(Box::new(reply));
189            None
190        }
191        Some((_, Taken::Answered(outcome))) => {
192            let outcome = outcome.clone();
193            drop(taken);
194            reply(outcome);
195            None
196        }
197        None => {
198            taken.insert(id, (Instant::now(), Taken::Running(vec![Box::new(reply)])));
199            Some(Pending {
200                id,
201                finished: false,
202            })
203        }
204    }
205}
206
207/// Answer a command that wants one through `reply`, and hand it on to be
208/// acted on, with what to finish, unless it is a repeat. Without an id it is
209/// acted on as it always was.
210pub fn take(
211    envelope: Envelope,
212    reply: impl FnOnce(u64, AckOutcome) + Send + 'static,
213) -> Option<(LinkCommand, Option<Pending>)> {
214    match envelope.ack {
215        None => Some((envelope.command, None)),
216        Some(id) => {
217            let reply = Arc::new(Mutex::new(Some(reply)));
218            let pending = accept(id, move |outcome| {
219                if let Some(reply) = reply.lock().take() {
220                    reply(id, outcome);
221                }
222            })?;
223            Some((envelope.command, Some(pending)))
224        }
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231
232    #[test]
233    fn an_envelope_is_the_command_with_an_id_beside_it() {
234        let pause = Envelope {
235            command: LinkCommand::Pause,
236            ack: Some(17),
237        };
238        let json = serde_json::to_string(&pause).unwrap();
239        assert_eq!(json, r#"{"type":"pause","ack":17}"#);
240        assert_eq!(serde_json::from_str::<Envelope>(&json).unwrap(), pause);
241        // A client that predates acks reads the command and ignores the id.
242        assert_eq!(
243            serde_json::from_str::<LinkCommand>(&json).unwrap(),
244            LinkCommand::Pause
245        );
246        // And a plain command is an envelope without one.
247        assert_eq!(
248            serde_json::from_str::<Envelope>(r#"{"type":"pause"}"#).unwrap(),
249            Envelope::from(LinkCommand::Pause)
250        );
251    }
252
253    #[test]
254    fn ids_are_random_not_counted() {
255        let ids: Vec<u64> = (0..8).map(|_| next_id()).collect();
256        assert!(ids.iter().all(|id| *id != 0));
257        let consecutive = ids.windows(2).filter(|w| w[1] == w[0] + 1).count();
258        assert_eq!(consecutive, 0, "one id says nothing of the next: {ids:?}");
259    }
260
261    #[test]
262    fn a_repeat_is_answered_with_the_first_outcome_and_not_acted_on() {
263        let id = next_id();
264        let answers = Arc::new(Mutex::new(Vec::new()));
265        let record = |n: u32| {
266            let answers = answers.clone();
267            move |outcome: AckOutcome| answers.lock().push((n, outcome))
268        };
269
270        let first = accept(id, record(1)).expect("acted on");
271        // Arriving by another way while the first is still running.
272        assert!(accept(id, record(2)).is_none(), "not acted on twice");
273        assert!(answers.lock().is_empty());
274
275        first.finish(AckOutcome::Done);
276        assert_eq!(
277            *answers.lock(),
278            [(1, AckOutcome::Done), (2, AckOutcome::Done)]
279        );
280
281        // And after it finished: answered at once.
282        assert!(accept(id, record(3)).is_none());
283        assert_eq!(answers.lock().last(), Some(&(3, AckOutcome::Done)));
284    }
285
286    #[test]
287    fn a_command_dropped_unfinished_is_answered_as_failed() {
288        let id = next_id();
289        let answer = Arc::new(Mutex::new(None));
290        let got = answer.clone();
291        drop(accept(id, move |o| *got.lock() = Some(o)));
292        assert!(matches!(*answer.lock(), Some(AckOutcome::Failed { .. })));
293    }
294
295    #[test]
296    fn a_sender_hears_its_answer_once() {
297        let id = next_id();
298        let rx = expect(id, "phone");
299        resolve(id, "phone", AckOutcome::Queued);
300        resolve(id, "phone", AckOutcome::Done);
301        assert_eq!(rx.try_recv(), Ok(AckOutcome::Queued));
302        assert!(rx.try_recv().is_err());
303    }
304
305    #[test]
306    fn only_the_device_a_command_went_to_answers_it() {
307        let id = next_id();
308        let rx = expect(id, "phone");
309        // Another device on the network, which saw the id go by.
310        resolve(id, "stranger", AckOutcome::Done);
311        assert!(rx.try_recv().is_err(), "a stranger's answer is not heard");
312        resolve(id, "phone", AckOutcome::Failed { error: "no".into() });
313        assert!(matches!(rx.try_recv(), Ok(AckOutcome::Failed { .. })));
314    }
315}
koan_core/remote/acks.rs

koan_core/remote/
acks.rs
