lonca_client/
client.rs

1//! Client for the daemon socket.
2
3use std::io::BufReader;
4use std::io::ErrorKind::{BrokenPipe, ConnectionReset, TimedOut, UnexpectedEof, WouldBlock};
5use std::os::unix::net::UnixStream;
6use std::path::{Path, PathBuf};
7use std::time::Duration;
8
9use serde::de::DeserializeOwned;
10use serde_json::Value;
11
12use crate::protocol::{ErrorBody, MAX_LINE, Request, Response, read_line_limited, write_line};
13use crate::trust::Trust;
14
15pub struct Client {
16    socket: PathBuf,
17    trust: Trust,
18    reader: BufReader<UnixStream>,
19    writer: UnixStream,
20    next: u64,
21    timeout: Option<Duration>,
22    /// The last exchange did not finish: what is still on the connection
23    /// could be the answer to it, so the next call opens a new one.
24    broken: bool,
25    pub token: Option<String>,
26    pub repo: Option<String>,
27}
28
29/// An error code of the daemon (`docs/protocol.md`, "Error codes"), or of
30/// this client: `io` (the connection failed) and `timed_out` (no answer
31/// within [`Client::set_timeout`]).
32#[derive(Clone, Copy, Debug, PartialEq, Eq)]
33#[non_exhaustive]
34pub enum ErrorKind {
35    Invalid,
36    Unauthenticated,
37    Forbidden,
38    NotInSandbox,
39    SessionInactive,
40    UnknownMethod,
41    NotFound,
42    TooLarge,
43    RateLimited,
44    Busy,
45    Verifying,
46    Closed,
47    /// The daemon's: the request line took too long to arrive.
48    Timeout,
49    Conflict,
50    Denied,
51    ForeignKey,
52    UpToDate,
53    NotFastForward,
54    Git,
55    Integrity,
56    Unsupported,
57    Internal,
58    /// The client's: the connection failed, or the answer was not one.
59    Io,
60    /// The client's: no answer within the timeout. The request may have
61    /// run; the connection is dropped and the next call opens another.
62    TimedOut,
63    /// A code this version does not know.
64    Other,
65}
66
67impl ErrorKind {
68    pub fn of(code: &str) -> Self {
69        use ErrorKind::*;
70        match code {
71            "invalid" => Invalid,
72            "unauthenticated" => Unauthenticated,
73            "forbidden" => Forbidden,
74            "not_in_sandbox" => NotInSandbox,
75            "session_inactive" => SessionInactive,
76            "unknown_method" => UnknownMethod,
77            "not_found" => NotFound,
78            "too_large" => TooLarge,
79            "rate_limited" => RateLimited,
80            "busy" => Busy,
81            "verifying" => Verifying,
82            "closed" => Closed,
83            "timeout" => Timeout,
84            "conflict" => Conflict,
85            "denied" => Denied,
86            "foreign_key" => ForeignKey,
87            "up_to_date" => UpToDate,
88            "not_fast_forward" => NotFastForward,
89            "git" => Git,
90            "integrity" => Integrity,
91            "unsupported" => Unsupported,
92            "internal" => Internal,
93            CODE_IO => Io,
94            CODE_TIMED_OUT => TimedOut,
95            _ => Other,
96        }
97    }
98
99    /// Nothing of the request ran, and the same request may be sent again:
100    /// `rate_limited` (after the wait its message names), `busy` and
101    /// `closed` (on a new connection). Everything else is an answer.
102    pub fn is_transient(self) -> bool {
103        matches!(
104            self,
105            ErrorKind::RateLimited | ErrorKind::Busy | ErrorKind::Closed
106        )
107    }
108}
109
110const CODE_IO: &str = "io";
111const CODE_TIMED_OUT: &str = "timed_out";
112
113#[derive(Debug)]
114pub struct CallError(pub ErrorBody);
115
116impl CallError {
117    pub fn code(&self) -> &str {
118        &self.0.code
119    }
120
121    pub fn kind(&self) -> ErrorKind {
122        ErrorKind::of(&self.0.code)
123    }
124
125    /// See [`ErrorKind::is_transient`].
126    pub fn is_transient(&self) -> bool {
127        self.kind().is_transient()
128    }
129}
130
131impl std::fmt::Display for CallError {
132    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133        write!(f, "{}: {}", self.0.code, self.0.message)
134    }
135}
136
137impl std::error::Error for CallError {}
138
139/// How a request and its answer did not go through.
140enum Exchange {
141    Write(std::io::Error),
142    Read(std::io::Error),
143    /// An answer with the id of another request.
144    Stray(u64),
145}
146
147impl From<Exchange> for CallError {
148    fn from(e: Exchange) -> Self {
149        match e {
150            Exchange::Write(e) | Exchange::Read(e) if matches!(e.kind(), WouldBlock | TimedOut) => {
151                error(CODE_TIMED_OUT, "no answer within the client's timeout")
152            }
153            Exchange::Write(e) | Exchange::Read(e) if e.kind() == UnexpectedEof => {
154                gone("daemon closed the connection")
155            }
156            Exchange::Write(e) | Exchange::Read(e) => io(e),
157            Exchange::Stray(id) => gone(&format!("an answer to request {id}, not this one")),
158        }
159    }
160}
161
162fn error(code: &str, message: &str) -> CallError {
163    CallError(ErrorBody {
164        code: code.into(),
165        message: message.into(),
166    })
167}
168
169fn gone(message: &str) -> CallError {
170    error(CODE_IO, message)
171}
172
173fn io(e: std::io::Error) -> CallError {
174    gone(&e.to_string())
175}
176
177/// The daemon had closed the connection before the request got there.
178fn not_sent(e: &Exchange) -> bool {
179    matches!(e, Exchange::Write(e) if matches!(e.kind(), BrokenPipe | ConnectionReset))
180}
181
182/// The daemon's last line on a connection it closed: id 0, which no request
183/// of this client has.
184fn not_read(r: &Response) -> bool {
185    r.id == 0 && r.error.as_ref().is_some_and(|e| e.code == "closed")
186}
187
188impl Client {
189    /// Connects to the daemon this machine trusts. There is no way to get
190    /// a client without the check: a token goes nowhere unverified.
191    pub fn connect(path: &Path) -> std::io::Result<Self> {
192        Self::connect_to(path, &Trust::load()?)
193    }
194
195    pub fn connect_to(path: &Path, trust: &Trust) -> std::io::Result<Self> {
196        trust.check_socket(path)?;
197        let s = UnixStream::connect(path)?;
198        trust.check_peer(&s)?;
199        Ok(Self {
200            socket: path.to_path_buf(),
201            trust: *trust,
202            reader: BufReader::new(s.try_clone()?),
203            writer: s,
204            next: 1,
205            timeout: None,
206            broken: false,
207            token: None,
208            repo: None,
209        })
210    }
211
212    /// How long a request may take to be sent and answered; `None` (the
213    /// default) waits as long as the daemon does. Mind `log`'s `wait_ms`.
214    /// Past it the call fails with `timed_out` and the connection is
215    /// dropped: the request may have run.
216    pub fn set_timeout(&mut self, timeout: Option<Duration>) -> std::io::Result<()> {
217        if timeout == Some(Duration::ZERO) {
218            return Err(std::io::Error::new(
219                std::io::ErrorKind::InvalidInput,
220                "a timeout of zero",
221            ));
222        }
223        self.timeout = timeout;
224        self.writer.set_read_timeout(timeout)?;
225        self.writer.set_write_timeout(timeout)
226    }
227
228    pub fn timeout(&self) -> Option<Duration> {
229        self.timeout
230    }
231
232    pub fn call(&mut self, method: &str, params: Value) -> Result<Value, CallError> {
233        if self.broken {
234            self.reconnect()?;
235        }
236        let id = self.next;
237        self.next += 1;
238        let req = Request {
239            id,
240            method: method.into(),
241            repo: self.repo.clone(),
242            token: self.token.clone(),
243            params,
244        };
245        let resp = match self.exchange(&req) {
246            Err(e) if not_sent(&e) => self.again(&req),
247            Ok(r) if not_read(&r) => self.again(&req),
248            done => done.map_err(CallError::from),
249        };
250        let resp = resp.inspect_err(|_| self.broken = true)?;
251        if resp.ok {
252            Ok(resp.result)
253        } else {
254            Err(CallError(resp.error.unwrap_or(ErrorBody {
255                code: "unknown".into(),
256                message: String::new(),
257            })))
258        }
259    }
260
261    fn reconnect(&mut self) -> Result<(), CallError> {
262        let mut fresh = Self::connect_to(&self.socket, &self.trust).map_err(io)?;
263        fresh.set_timeout(self.timeout).map_err(io)?;
264        self.reader = fresh.reader;
265        self.writer = fresh.writer;
266        self.broken = false;
267        Ok(())
268    }
269
270    /// The daemon closed this connection for being quiet and never saw the
271    /// request, so it is safe to send on a new one, once.
272    fn again(&mut self, req: &Request) -> Result<Response, CallError> {
273        self.reconnect()?;
274        Ok(self.exchange(req)?)
275    }
276
277    fn exchange(&mut self, req: &Request) -> Result<Response, Exchange> {
278        write_line(&mut self.writer, req).map_err(Exchange::Write)?;
279        let line = read_line_limited(&mut self.reader, MAX_LINE)
280            .map_err(Exchange::Read)?
281            .ok_or_else(|| Exchange::Read(UnexpectedEof.into()))?;
282        let resp: Response = serde_json::from_str(&line).map_err(|e| Exchange::Read(e.into()))?;
283        // Id 0 is the daemon's word on the connection, not on a request.
284        if resp.id != req.id && resp.id != 0 {
285            return Err(Exchange::Stray(resp.id));
286        }
287        Ok(resp)
288    }
289
290    pub fn call_as<T: DeserializeOwned>(
291        &mut self,
292        method: &str,
293        params: Value,
294    ) -> Result<T, CallError> {
295        let v = self.call(method, params)?;
296        serde_json::from_value(v).map_err(|e| gone(&format!("bad result: {e}")))
297    }
298
299    /// `events.watch` with the human session this client carries: what
300    /// happened to its agents' sessions, held up to `watch.wait_ms` while
301    /// nothing did. Go on with `watch.after(&answer)`. A client timeout
302    /// ([`Client::set_timeout`]) must be longer than the wait.
303    pub fn watch(
304        &mut self,
305        watch: &crate::protocol::Watch,
306    ) -> Result<crate::protocol::Watched, CallError> {
307        let params = serde_json::to_value(watch).map_err(|e| gone(&format!("bad watch: {e}")))?;
308        self.call_as("events.watch", params)
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315    use crate::protocol::write_line;
316    use std::io::BufRead;
317    use std::os::unix::net::UnixListener;
318
319    /// A daemon of one connection at a time: `answer` gets each request
320    /// line and says what to send back (nothing: stay quiet).
321    fn fake(
322        answer: impl Fn(usize, &Request) -> Option<Value> + Send + Sync + 'static,
323    ) -> (tempfile::TempDir, PathBuf) {
324        let dir = tempfile::Builder::new().prefix("lc").tempdir().unwrap();
325        let sock = dir.path().join("s");
326        let listener = UnixListener::bind(&sock).unwrap();
327        let answer = std::sync::Arc::new(answer);
328        let count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
329        std::thread::spawn(move || {
330            for conn in listener.incoming() {
331                let Ok(conn) = conn else { return };
332                let (answer, count) = (answer.clone(), count.clone());
333                std::thread::spawn(move || {
334                    let mut reader = std::io::BufReader::new(conn.try_clone().unwrap());
335                    let mut w = conn;
336                    let mut line = String::new();
337                    while reader.read_line(&mut line).unwrap_or(0) > 0 {
338                        let req: Request = serde_json::from_str(&line).unwrap();
339                        line.clear();
340                        let n = count.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
341                        if let Some(v) = answer(n, &req) {
342                            let _ = write_line(&mut w, &v);
343                        }
344                    }
345                });
346            }
347        });
348        (dir, sock)
349    }
350
351    fn client(sock: &Path) -> Client {
352        Client::connect_to(sock, &Trust::cooperative()).unwrap()
353    }
354
355    #[test]
356    fn every_code_of_the_table_has_its_kind() {
357        let codes = [
358            ("invalid", ErrorKind::Invalid),
359            ("unauthenticated", ErrorKind::Unauthenticated),
360            ("forbidden", ErrorKind::Forbidden),
361            ("not_in_sandbox", ErrorKind::NotInSandbox),
362            ("session_inactive", ErrorKind::SessionInactive),
363            ("unknown_method", ErrorKind::UnknownMethod),
364            ("not_found", ErrorKind::NotFound),
365            ("too_large", ErrorKind::TooLarge),
366            ("rate_limited", ErrorKind::RateLimited),
367            ("busy", ErrorKind::Busy),
368            ("verifying", ErrorKind::Verifying),
369            ("closed", ErrorKind::Closed),
370            ("timeout", ErrorKind::Timeout),
371            ("conflict", ErrorKind::Conflict),
372            ("denied", ErrorKind::Denied),
373            ("foreign_key", ErrorKind::ForeignKey),
374            ("up_to_date", ErrorKind::UpToDate),
375            ("not_fast_forward", ErrorKind::NotFastForward),
376            ("git", ErrorKind::Git),
377            ("integrity", ErrorKind::Integrity),
378            ("unsupported", ErrorKind::Unsupported),
379            ("internal", ErrorKind::Internal),
380            ("io", ErrorKind::Io),
381            ("timed_out", ErrorKind::TimedOut),
382            ("something_new", ErrorKind::Other),
383        ];
384        for (code, kind) in codes {
385            assert_eq!(ErrorKind::of(code), kind, "{code}");
386            let transient = matches!(code, "rate_limited" | "busy" | "closed");
387            assert_eq!(kind.is_transient(), transient, "{code}");
388            let e = error(code, "m");
389            assert_eq!(
390                (e.code(), e.kind(), e.is_transient()),
391                (code, kind, transient)
392            );
393            assert_eq!(e.to_string(), format!("{code}: m"));
394        }
395    }
396
397    #[test]
398    fn a_call_gets_its_result_or_the_daemons_error() {
399        let (_d, sock) = fake(|_, req| {
400            Some(match req.method.as_str() {
401                "ping" => serde_json::json!({"id": req.id, "ok": true, "result": {"protocol": 1}}),
402                _ => serde_json::json!({"id": req.id, "ok": false,
403                    "error": {"code": "rate_limited", "message": "try again in 3 s"}}),
404            })
405        });
406        let mut c = client(&sock);
407        c.token = Some("lonca_t".into());
408        c.repo = Some("demo".into());
409        let v = c.call("ping", serde_json::json!({})).unwrap();
410        assert_eq!(v["protocol"], crate::PROTOCOL);
411        #[derive(serde::Deserialize)]
412        struct Ping {
413            protocol: u32,
414        }
415        let p: Ping = c.call_as("ping", serde_json::json!({})).unwrap();
416        assert_eq!(p.protocol, 1);
417        let bad: Result<Vec<u8>, _> = c.call_as("ping", serde_json::json!({}));
418        assert_eq!(bad.unwrap_err().kind(), ErrorKind::Io);
419        let e = c.call("status", serde_json::json!({})).unwrap_err();
420        assert_eq!(e.kind(), ErrorKind::RateLimited);
421        // Each request has an id of its own, counting up.
422        let (_d, sock) = fake(|_, req| {
423            Some(serde_json::json!({"id": req.id, "ok": true,
424                "result": [req.id, req.token, req.repo]}))
425        });
426        let mut c = client(&sock);
427        c.token = Some("lonca_t".into());
428        let first = c.call("ping", serde_json::json!({})).unwrap();
429        assert_eq!(first, serde_json::json!([1, "lonca_t", null]));
430        c.repo = Some("demo".into());
431        let second = c.call("ping", serde_json::json!({})).unwrap();
432        assert_eq!(second, serde_json::json!([2, "lonca_t", "demo"]));
433        assert!(e.is_transient());
434        assert!(e.to_string().contains("try again"), "{e}");
435    }
436
437    /// An error with no body is still an error.
438    #[test]
439    fn an_error_without_a_body_is_unknown() {
440        let (_d, sock) = fake(|_, req| Some(serde_json::json!({"id": req.id, "ok": false})));
441        let e = client(&sock)
442            .call("ping", serde_json::json!({}))
443            .unwrap_err();
444        assert_eq!((e.code(), e.kind()), ("unknown", ErrorKind::Other));
445    }
446
447    /// The request and the answer carry the same id; another id is not an
448    /// answer, and the connection is not used again.
449    #[test]
450    fn an_answer_to_another_request_is_refused() {
451        let (_d, sock) = fake(|n, req| {
452            let id = if n == 1 { req.id + 7 } else { req.id };
453            Some(serde_json::json!({"id": id, "ok": true, "result": n}))
454        });
455        let mut c = client(&sock);
456        let e = c.call("ping", serde_json::json!({})).unwrap_err();
457        assert_eq!(e.kind(), ErrorKind::Io);
458        assert!(e.to_string().contains("request 8"), "{e}");
459        // A new connection, and the next answer is its own.
460        assert_eq!(c.call("ping", serde_json::json!({})).unwrap(), 2);
461    }
462
463    /// No answer in time: `timed_out`, and the late answer is never taken
464    /// for the next request's.
465    #[test]
466    fn a_timeout_drops_the_connection() {
467        let (_d, sock) = fake(|n, req| match n {
468            1 => {
469                std::thread::sleep(Duration::from_millis(400));
470                Some(serde_json::json!({"id": req.id, "ok": true, "result": "late"}))
471            }
472            _ => Some(serde_json::json!({"id": req.id, "ok": true, "result": "fresh"})),
473        });
474        let mut c = client(&sock);
475        assert!(c.set_timeout(Some(Duration::ZERO)).is_err());
476        assert_eq!(c.timeout(), None);
477        c.set_timeout(Some(Duration::from_millis(100))).unwrap();
478        assert_eq!(c.timeout(), Some(Duration::from_millis(100)));
479        let e = c.call("ping", serde_json::json!({})).unwrap_err();
480        assert_eq!(e.kind(), ErrorKind::TimedOut, "{e}");
481        assert!(!e.is_transient(), "it may have run");
482        assert_eq!(c.call("ping", serde_json::json!({})).unwrap(), "fresh");
483        assert_eq!(c.timeout(), Some(Duration::from_millis(100)), "kept");
484    }
485
486    /// The daemon's farewell on a quiet connection: the request was never
487    /// read, so it goes again on a new connection, once.
488    #[test]
489    fn a_closed_connection_is_sent_again_once() {
490        let (_d, sock) = fake(|n, req| {
491            Some(match n {
492                1 | 3 | 4 => serde_json::json!({"id": 0, "ok": false,
493                    "error": {"code": "closed", "message": "quiet"}}),
494                _ => serde_json::json!({"id": req.id, "ok": true, "result": n}),
495            })
496        });
497        let mut c = client(&sock);
498        assert_eq!(c.call("ping", serde_json::json!({})).unwrap(), 2);
499        let e = c.call("ping", serde_json::json!({})).unwrap_err();
500        assert_eq!(e.kind(), ErrorKind::Closed, "not twice");
501        assert!(e.is_transient());
502    }
503
504    #[test]
505    fn a_closed_socket_is_an_io_error() {
506        let (_d, sock) = fake(|_, _| None);
507        let mut c = client(&sock);
508        c.writer.shutdown(std::net::Shutdown::Read).unwrap();
509        let e = c.call("ping", serde_json::json!({})).unwrap_err();
510        assert_eq!(e.kind(), ErrorKind::Io, "{e}");
511    }
512}
lonca_client/client.rs

lonca_client/
client.rs
