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
300#[cfg(test)]
301mod tests {
302    use super::*;
303    use crate::protocol::write_line;
304    use std::io::BufRead;
305    use std::os::unix::net::UnixListener;
306
307    /// A daemon of one connection at a time: `answer` gets each request
308    /// line and says what to send back (nothing: stay quiet).
309    fn fake(
310        answer: impl Fn(usize, &Request) -> Option<Value> + Send + Sync + 'static,
311    ) -> (tempfile::TempDir, PathBuf) {
312        let dir = tempfile::Builder::new().prefix("lc").tempdir().unwrap();
313        let sock = dir.path().join("s");
314        let listener = UnixListener::bind(&sock).unwrap();
315        let answer = std::sync::Arc::new(answer);
316        let count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
317        std::thread::spawn(move || {
318            for conn in listener.incoming() {
319                let Ok(conn) = conn else { return };
320                let (answer, count) = (answer.clone(), count.clone());
321                std::thread::spawn(move || {
322                    let mut reader = std::io::BufReader::new(conn.try_clone().unwrap());
323                    let mut w = conn;
324                    let mut line = String::new();
325                    while reader.read_line(&mut line).unwrap_or(0) > 0 {
326                        let req: Request = serde_json::from_str(&line).unwrap();
327                        line.clear();
328                        let n = count.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
329                        if let Some(v) = answer(n, &req) {
330                            let _ = write_line(&mut w, &v);
331                        }
332                    }
333                });
334            }
335        });
336        (dir, sock)
337    }
338
339    fn client(sock: &Path) -> Client {
340        Client::connect_to(sock, &Trust::cooperative()).unwrap()
341    }
342
343    #[test]
344    fn every_code_of_the_table_has_its_kind() {
345        let codes = [
346            ("invalid", ErrorKind::Invalid),
347            ("unauthenticated", ErrorKind::Unauthenticated),
348            ("forbidden", ErrorKind::Forbidden),
349            ("not_in_sandbox", ErrorKind::NotInSandbox),
350            ("session_inactive", ErrorKind::SessionInactive),
351            ("unknown_method", ErrorKind::UnknownMethod),
352            ("not_found", ErrorKind::NotFound),
353            ("too_large", ErrorKind::TooLarge),
354            ("rate_limited", ErrorKind::RateLimited),
355            ("busy", ErrorKind::Busy),
356            ("verifying", ErrorKind::Verifying),
357            ("closed", ErrorKind::Closed),
358            ("timeout", ErrorKind::Timeout),
359            ("conflict", ErrorKind::Conflict),
360            ("denied", ErrorKind::Denied),
361            ("foreign_key", ErrorKind::ForeignKey),
362            ("up_to_date", ErrorKind::UpToDate),
363            ("not_fast_forward", ErrorKind::NotFastForward),
364            ("git", ErrorKind::Git),
365            ("integrity", ErrorKind::Integrity),
366            ("unsupported", ErrorKind::Unsupported),
367            ("internal", ErrorKind::Internal),
368            ("io", ErrorKind::Io),
369            ("timed_out", ErrorKind::TimedOut),
370            ("something_new", ErrorKind::Other),
371        ];
372        for (code, kind) in codes {
373            assert_eq!(ErrorKind::of(code), kind, "{code}");
374            let transient = matches!(code, "rate_limited" | "busy" | "closed");
375            assert_eq!(kind.is_transient(), transient, "{code}");
376            let e = error(code, "m");
377            assert_eq!(
378                (e.code(), e.kind(), e.is_transient()),
379                (code, kind, transient)
380            );
381            assert_eq!(e.to_string(), format!("{code}: m"));
382        }
383    }
384
385    #[test]
386    fn a_call_gets_its_result_or_the_daemons_error() {
387        let (_d, sock) = fake(|_, req| {
388            Some(match req.method.as_str() {
389                "ping" => serde_json::json!({"id": req.id, "ok": true, "result": {"protocol": 1}}),
390                _ => serde_json::json!({"id": req.id, "ok": false,
391                    "error": {"code": "rate_limited", "message": "try again in 3 s"}}),
392            })
393        });
394        let mut c = client(&sock);
395        c.token = Some("lonca_t".into());
396        c.repo = Some("demo".into());
397        let v = c.call("ping", serde_json::json!({})).unwrap();
398        assert_eq!(v["protocol"], crate::PROTOCOL);
399        #[derive(serde::Deserialize)]
400        struct Ping {
401            protocol: u32,
402        }
403        let p: Ping = c.call_as("ping", serde_json::json!({})).unwrap();
404        assert_eq!(p.protocol, 1);
405        let bad: Result<Vec<u8>, _> = c.call_as("ping", serde_json::json!({}));
406        assert_eq!(bad.unwrap_err().kind(), ErrorKind::Io);
407        let e = c.call("status", serde_json::json!({})).unwrap_err();
408        assert_eq!(e.kind(), ErrorKind::RateLimited);
409        // Each request has an id of its own, counting up.
410        let (_d, sock) = fake(|_, req| {
411            Some(serde_json::json!({"id": req.id, "ok": true,
412                "result": [req.id, req.token, req.repo]}))
413        });
414        let mut c = client(&sock);
415        c.token = Some("lonca_t".into());
416        let first = c.call("ping", serde_json::json!({})).unwrap();
417        assert_eq!(first, serde_json::json!([1, "lonca_t", null]));
418        c.repo = Some("demo".into());
419        let second = c.call("ping", serde_json::json!({})).unwrap();
420        assert_eq!(second, serde_json::json!([2, "lonca_t", "demo"]));
421        assert!(e.is_transient());
422        assert!(e.to_string().contains("try again"), "{e}");
423    }
424
425    /// An error with no body is still an error.
426    #[test]
427    fn an_error_without_a_body_is_unknown() {
428        let (_d, sock) = fake(|_, req| Some(serde_json::json!({"id": req.id, "ok": false})));
429        let e = client(&sock)
430            .call("ping", serde_json::json!({}))
431            .unwrap_err();
432        assert_eq!((e.code(), e.kind()), ("unknown", ErrorKind::Other));
433    }
434
435    /// The request and the answer carry the same id; another id is not an
436    /// answer, and the connection is not used again.
437    #[test]
438    fn an_answer_to_another_request_is_refused() {
439        let (_d, sock) = fake(|n, req| {
440            let id = if n == 1 { req.id + 7 } else { req.id };
441            Some(serde_json::json!({"id": id, "ok": true, "result": n}))
442        });
443        let mut c = client(&sock);
444        let e = c.call("ping", serde_json::json!({})).unwrap_err();
445        assert_eq!(e.kind(), ErrorKind::Io);
446        assert!(e.to_string().contains("request 8"), "{e}");
447        // A new connection, and the next answer is its own.
448        assert_eq!(c.call("ping", serde_json::json!({})).unwrap(), 2);
449    }
450
451    /// No answer in time: `timed_out`, and the late answer is never taken
452    /// for the next request's.
453    #[test]
454    fn a_timeout_drops_the_connection() {
455        let (_d, sock) = fake(|n, req| match n {
456            1 => {
457                std::thread::sleep(Duration::from_millis(400));
458                Some(serde_json::json!({"id": req.id, "ok": true, "result": "late"}))
459            }
460            _ => Some(serde_json::json!({"id": req.id, "ok": true, "result": "fresh"})),
461        });
462        let mut c = client(&sock);
463        assert!(c.set_timeout(Some(Duration::ZERO)).is_err());
464        assert_eq!(c.timeout(), None);
465        c.set_timeout(Some(Duration::from_millis(100))).unwrap();
466        assert_eq!(c.timeout(), Some(Duration::from_millis(100)));
467        let e = c.call("ping", serde_json::json!({})).unwrap_err();
468        assert_eq!(e.kind(), ErrorKind::TimedOut, "{e}");
469        assert!(!e.is_transient(), "it may have run");
470        assert_eq!(c.call("ping", serde_json::json!({})).unwrap(), "fresh");
471        assert_eq!(c.timeout(), Some(Duration::from_millis(100)), "kept");
472    }
473
474    /// The daemon's farewell on a quiet connection: the request was never
475    /// read, so it goes again on a new connection, once.
476    #[test]
477    fn a_closed_connection_is_sent_again_once() {
478        let (_d, sock) = fake(|n, req| {
479            Some(match n {
480                1 | 3 | 4 => serde_json::json!({"id": 0, "ok": false,
481                    "error": {"code": "closed", "message": "quiet"}}),
482                _ => serde_json::json!({"id": req.id, "ok": true, "result": n}),
483            })
484        });
485        let mut c = client(&sock);
486        assert_eq!(c.call("ping", serde_json::json!({})).unwrap(), 2);
487        let e = c.call("ping", serde_json::json!({})).unwrap_err();
488        assert_eq!(e.kind(), ErrorKind::Closed, "not twice");
489        assert!(e.is_transient());
490    }
491
492    #[test]
493    fn a_closed_socket_is_an_io_error() {
494        let (_d, sock) = fake(|_, _| None);
495        let mut c = client(&sock);
496        c.writer.shutdown(std::net::Shutdown::Read).unwrap();
497        let e = c.call("ping", serde_json::json!({})).unwrap_err();
498        assert_eq!(e.kind(), ErrorKind::Io, "{e}");
499    }
500}
lonca_client/client.rs

lonca_client/
client.rs
