Skip to main content

haematite_cli/
client.rs

1//! `haematite call`: ask a running service anything its protocol offers, from the
2//! command line; `haematite write` is in [`write`] and `haematite query` in [`query`].
3//!
4//! The client says hello first and takes the service's `max_frame` as the
5//! bound on what it sends and what it reads. Params go out as the text given
6//! and a result comes back as the bytes the service sent: no number is parsed
7//! into a float and written again. Four outcomes are told apart by exit code:
8//! answered (0), could not ask, so nothing was sent (1), refused by the service
9//! (2), and outcome not known to the caller (3).
10
11use crate::serve::protocol::{HelloAgreement, PROTOCOL, RECEIPT_CAPABILITY};
12use crate::serve::receipt_fragments::{Assembler, VerifiedReceipt, minimum_cap};
13use std::io::{self, Read, Write};
14use std::num::{NonZeroU32, NonZeroU64};
15use std::os::unix::net::UnixStream;
16use std::path::{Path, PathBuf};
17use std::process::ExitCode;
18use std::sync::{Arc, mpsc};
19use std::time::Duration;
20
21use serde::{Deserialize, Deserializer, Serialize};
22use serde_json::value::RawValue;
23
24use crate::serve::{cli::Config, frame};
25
26pub mod query;
27pub mod write;
28
29const USAGE: &str = "\
30usage: haematite call (--socket <path> | --config <serve.json>) [--timeout-ms N] <method> [<params> | -]
31<params> is one JSON value; - reads it from standard input; absent, the operation has no params.
32--timeout-ms bounds connecting, writing and reading; the default is 30000.
33Exit 0: answered, the result on standard output. Exit 1: could not ask; nothing was sent.
34Exit 2: refused by the service, its code and message on standard error. Exit 3: the outcome
35is not known to the caller; the service may have acted.";
36
37const DEFAULT_TIMEOUT_MS: NonZeroU64 = NonZeroU64::MIN.saturating_add(29_999);
38/// Hello's answer is read before the service has said its frame limit.
39const HELLO_BOUND: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
40const ID: &str = "haem-call";
41
42/// Codes which leave the mutation's outcome uncertain.
43///
44/// Unrecognised codes also retain uncertainty unless the shared refusal table
45/// proves otherwise.
46pub const MAY_HAVE_APPLIED: &[&str] = &[
47    // The commit failed after the branch reference moved: storage/mutations.rs
48    // 196 and 476 (a fork); a discard, storage/discards.rs 226, 296, 332, 418
49    // and 424; a receipt's confirmation, storage/mutations.rs 71.
50    "durability_unknown",
51    // The change is durable and the catalogue could not show it:
52    // storage/catalogue.rs 37, by publish_record after every commit.
53    "committed_unpublished",
54    // The reference vanished after its commit: storage.rs 275, the same
55    // publish_record; it is also raised before commits, which cannot be told
56    // apart from here.
57    "reference_missing",
58    // A refused discard commit whose intent could not be withdrawn:
59    // storage/discards.rs 253.
60    "discard_abort_failed",
61    // The catalogue could not hide a branch whose discard intent was
62    // installed: storage/catalogue.rs 49, called from storage/discards.rs 221
63    // and after the commit at 416, 425 and 429.
64    "catalogue_unavailable",
65    // Raised by Failure::storage everywhere, including after the change, as
66    // when the receipt cannot be serialized (serve/service.rs 123-124).
67    "storage_unavailable",
68    // The answer did not fit after the change was made: serve/service.rs 108
69    // and serve/transport.rs 177, which says to reconcile by request id.
70    "response_limit",
71];
72
73/// The service's shared proof that a refusal applied no mutation.
74pub use crate::serve::protocol::NOT_APPLIED_REFUSALS as KNOWN_REFUSALS;
75
76/// Why an exchange gave neither an answer nor a refusal.
77#[derive(Debug)]
78pub enum Ended {
79    /// The command line was not one the command takes; the usage text follows.
80    Usage(String),
81    /// Nothing the service could act on was sent.
82    NotAsked(String),
83    /// The request was sent, or may have been; whether the service acted on
84    /// it is not known.
85    Unknown(String),
86    /// The service sent a complete, verified receipt, and whether the change
87    /// was applied is still not known: the receipt carried no refusal, or its
88    /// code is one that may have applied or that this client does not know.
89    /// The receipt is kept as the exact bytes sent, to reconcile by.
90    Received {
91        why: String,
92        receipt: Arc<VerifiedReceipt>,
93    },
94}
95
96impl Ended {
97    /// The cause, without the words for which kind of end it was.
98    #[must_use]
99    pub fn why(&self) -> &str {
100        match self {
101            Self::Usage(why)
102            | Self::NotAsked(why)
103            | Self::Unknown(why)
104            | Self::Received { why, .. } => why,
105        }
106    }
107}
108
109impl std::fmt::Display for Ended {
110    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        let kind = match self {
112            Self::Usage(_) => "usage",
113            Self::NotAsked(_) => "not asked",
114            Self::Unknown(_) => "outcome unknown",
115            Self::Received { .. } => "complete receipt received",
116        };
117        write!(formatter, "{kind}: {}", self.why())
118    }
119}
120
121impl std::error::Error for Ended {}
122
123/// What the service said to one request.
124#[derive(Debug)]
125pub enum Reply {
126    /// The bytes the service sent for `result`.
127    Answered(Box<RawValue>),
128    Refused(Refusal),
129}
130
131/// A refusal by name, as the service sent it.
132#[derive(Debug, Deserialize)]
133pub struct Refusal {
134    pub code: String,
135    pub message: String,
136    #[serde(default)]
137    pub path: Option<String>,
138    #[serde(skip)]
139    pub verified: Option<Arc<VerifiedReceipt>>,
140}
141
142/// One connection to a running service, past its hello.
143pub struct Client {
144    stream: UnixStream,
145    hello: Box<RawValue>,
146    max_frame: NonZeroU32,
147    max_page: NonZeroU32,
148}
149
150#[derive(Deserialize)]
151struct Limits {
152    protocol: String,
153    capabilities: Vec<String>,
154    max_frame: NonZeroU32,
155    max_page: NonZeroU32,
156}
157
158impl Client {
159    /// Connect to `socket` and say hello, each step bounded by `timeout`.
160    ///
161    /// # Errors
162    ///
163    /// `Ended::NotAsked` naming the cause: hello changes nothing, so any
164    /// failure up to its answer leaves nothing asked.
165    pub fn connect(socket: &Path, timeout: Duration) -> Result<Self, Ended> {
166        let mut stream = connect(socket, timeout)?;
167        let unset =
168            |error: io::Error| Ended::NotAsked(format!("the timeout could not be set: {error}"));
169        stream.set_read_timeout(Some(timeout)).map_err(unset)?;
170        stream.set_write_timeout(Some(timeout)).map_err(unset)?;
171        let agreement = serde_json::value::to_raw_value(&HelloAgreement::client())
172            .map_err(|error| Ended::NotAsked(format!("receipt_hello_encode_failed: {error}")))?;
173        let hello = match exchange(&mut stream, HELLO_BOUND, "hello", Some(&agreement)) {
174            Ok(Reply::Answered(hello)) => hello,
175            Ok(Reply::Refused(refusal)) => {
176                return Err(Ended::NotAsked(format!(
177                    "the service refused hello: {}: {}",
178                    refusal.code, refusal.message
179                )));
180            }
181            Err(Ended::Received { why, receipt }) => {
182                return Err(Ended::Received {
183                    why: format!("hello failed: {why}"),
184                    receipt,
185                });
186            }
187            Err(Ended::Unknown(why) | Ended::NotAsked(why) | Ended::Usage(why)) => {
188                return Err(Ended::NotAsked(format!("hello failed: {why}")));
189            }
190        };
191        let limits: Limits = serde_json::from_str(hello.get()).map_err(|error| {
192            Ended::NotAsked(format!(
193                "the service's hello did not give its limits: {error}"
194            ))
195        })?;
196        if limits.protocol != PROTOCOL {
197            return Err(Ended::NotAsked(format!(
198                "receipt_protocol_mismatch: expected {PROTOCOL}, received {}",
199                limits.protocol
200            )));
201        }
202        if !limits
203            .capabilities
204            .iter()
205            .any(|capability| capability == RECEIPT_CAPABILITY)
206        {
207            return Err(Ended::NotAsked(format!(
208                "typed_receipt_capability_missing: required {RECEIPT_CAPABILITY}"
209            )));
210        }
211        let minimum = minimum_cap().map_err(|error| Ended::NotAsked(error.to_string()))?;
212        if u64::from(limits.max_frame.get())
213            < u64::try_from(minimum).map_err(|error| Ended::NotAsked(error.to_string()))?
214        {
215            return Err(Ended::NotAsked(format!(
216                "receipt_frame_too_small: offered {}, minimum {minimum}",
217                limits.max_frame
218            )));
219        }
220        Ok(Self {
221            stream,
222            hello,
223            max_frame: limits.max_frame,
224            max_page: limits.max_page,
225        })
226    }
227
228    /// The bytes the service sent for hello's `result`.
229    #[must_use]
230    pub fn hello(&self) -> &RawValue {
231        &self.hello
232    }
233
234    /// The largest frame the service reads or sends.
235    #[must_use]
236    pub const fn max_frame(&self) -> NonZeroU32 {
237        self.max_frame
238    }
239
240    /// The most rows the service puts on one page.
241    #[must_use]
242    pub const fn max_page(&self) -> NonZeroU32 {
243        self.max_page
244    }
245
246    /// Send one request and read its reply.
247    ///
248    /// # Errors
249    ///
250    /// `Ended::NotAsked` for a request larger than the service reads, refused
251    /// before sending; `Ended::Unknown` for any failure from sending onward,
252    /// and for a `mutation.apply` error that is not one of
253    /// [`KNOWN_REFUSALS`]: those the service sends when the change may already
254    /// have been applied ([`MAY_HAVE_APPLIED`]), and any code this client does
255    /// not know.
256    pub fn call(&mut self, method: &str, params: Option<&RawValue>) -> Result<Reply, Ended> {
257        let reply = exchange(&mut self.stream, self.max_frame, method, params)?;
258        let Reply::Refused(refusal) = reply else {
259            return Ok(reply);
260        };
261        // A read changes nothing, so a refusal of a read is a refusal.
262        if method != "mutation.apply" || KNOWN_REFUSALS.contains(&refusal.code.as_str()) {
263            return Ok(Reply::Refused(refusal));
264        }
265        let why = if MAY_HAVE_APPLIED.contains(&refusal.code.as_str()) {
266            "the change may have been applied"
267        } else {
268            "this client does not know that code, so whether the change was applied is not known"
269        };
270        let why = format!(
271            "the service answered {}: {}; {why}",
272            refusal.code, refusal.message
273        );
274        match refusal.verified {
275            Some(receipt) => Err(Ended::Received { why, receipt }),
276            None => Err(Ended::Unknown(why)),
277        }
278    }
279
280    /// Read the next frame of an answer that is a stream, such as
281    /// `head.follow`'s, sent under the same id as its first. A read changes
282    /// nothing, so a refusal is a refusal.
283    ///
284    /// # Errors
285    ///
286    /// `Ended::Unknown` when the frame could not be read: the connection
287    /// closed, failed or timed out.
288    pub fn next_reply(&mut self) -> Result<Reply, Ended> {
289        receive(&mut self.stream, self.max_frame)
290    }
291
292    /// An owned socket handle used only to cancel a disconnected browser's follow.
293    pub(crate) fn follow_socket(&self) -> io::Result<UnixStream> {
294        self.stream.try_clone()
295    }
296
297    /// Bound each later read by `timeout`, or by nothing with `None`.
298    ///
299    /// # Errors
300    ///
301    /// `Ended::Unknown` when the bound could not be set.
302    pub fn read_within(&self, timeout: Option<Duration>) -> Result<(), Ended> {
303        self.stream
304            .set_read_timeout(timeout)
305            .map_err(|error| Ended::Unknown(format!("the read bound could not be set: {error}")))
306    }
307}
308
309/// `haematite call`, ending in its exit code.
310#[must_use]
311pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
312    finish(call(args))
313}
314
315/// Print what an exchange came to and give its exit code: 0 answered, 1 could
316/// not ask, 2 refused, 3 outcome not known to the caller.
317fn finish(outcome: Result<Reply, Ended>) -> ExitCode {
318    match outcome {
319        Ok(Reply::Answered(result)) => {
320            let mut output = io::stdout().lock();
321            match writeln!(output, "{}", result.get()).and_then(|()| output.flush()) {
322                Ok(()) => ExitCode::SUCCESS,
323                Err(error) => {
324                    // The service acted and the caller did not get its answer.
325                    eprintln!(
326                        "haematite: answer received, but the result could not be written to stdout: {error}"
327                    );
328                    ExitCode::from(3)
329                }
330            }
331        }
332        Ok(Reply::Refused(refusal)) => {
333            if let Err(error) = print_refusal(&refusal) {
334                eprintln!("haematite: receipt_output_failed: {error}");
335            }
336            ExitCode::from(2)
337        }
338        Err(ended) => {
339            if let Err(error) = print_ended(&ended) {
340                eprintln!("haematite: receipt_output_failed: {error}");
341            }
342            ExitCode::from(exit_code(&ended))
343        }
344    }
345}
346
347/// 1 when nothing was sent, so asking again is safe; 3 when it may have been.
348#[must_use]
349pub const fn exit_code(ended: &Ended) -> u8 {
350    match ended {
351        Ended::Usage(_) | Ended::NotAsked(_) => 1,
352        Ended::Unknown(_) | Ended::Received { .. } => 3,
353    }
354}
355
356/// The standard error line for a refusal.
357#[must_use]
358pub fn refused(refusal: &Refusal) -> String {
359    let at = refusal
360        .path
361        .as_deref()
362        .map_or_else(String::new, |path| format!(" at {path}"));
363    let (code, message) = (&refusal.code, &refusal.message);
364    format!("haematite: refused {code}: {message}{at}")
365}
366
367/// The standard error line for an exchange that gave no reply.
368#[must_use]
369pub fn words(ended: &Ended) -> String {
370    match ended {
371        Ended::Usage(why) => format!("haematite: {why}"),
372        Ended::NotAsked(why) => format!("haematite: could not ask: {why}"),
373        Ended::Unknown(why) => format!("haematite: the outcome is not known: {why}"),
374        Ended::Received { why, .. } => format!("haematite: complete receipt received: {why}"),
375    }
376}
377
378pub(super) fn print_refusal(refusal: &Refusal) -> io::Result<()> {
379    let mut output = io::stderr().lock();
380    writeln!(output, "{}", refused(refusal))?;
381    if let Some(receipt) = &refusal.verified {
382        output.write_all(receipt.bytes())?;
383        output.write_all(b"\n")?;
384    }
385    output.flush()
386}
387pub(super) fn print_ended(ended: &Ended) -> io::Result<()> {
388    let mut output = io::stderr().lock();
389    writeln!(output, "{}", words(ended))?;
390    if let Ended::Received { receipt, .. } = ended {
391        output.write_all(receipt.bytes())?;
392        output.write_all(b"\n")?;
393    }
394    output.flush()
395}
396
397enum Target {
398    Socket(PathBuf),
399    Config(PathBuf),
400}
401
402/// The flags every client command takes: where the service is, and how long
403/// to wait for it.
404#[derive(Default)]
405struct Common {
406    target: Option<Target>,
407    timeout: Option<NonZeroU64>,
408}
409
410impl Common {
411    /// Take `arg`, and its value from `args`, when it is a shared flag.
412    fn take(&mut self, arg: &str, args: &mut impl Iterator<Item = String>) -> Result<bool, String> {
413        match arg {
414            "--socket" | "--config" if self.target.is_none() => {
415                let path = PathBuf::from(args.next().ok_or_else(|| format!("{arg} needs a path"))?);
416                self.target = Some(if arg == "--socket" {
417                    Target::Socket(path)
418                } else {
419                    Target::Config(path)
420                });
421                Ok(true)
422            }
423            "--socket" | "--config" => Err("give exactly one of --socket and --config".to_owned()),
424            "--timeout-ms" if self.timeout.is_none() => {
425                let text = args
426                    .next()
427                    .ok_or("--timeout-ms needs a number of milliseconds")?;
428                let millis = text
429                    .parse::<NonZeroU64>()
430                    .map_err(|error| format!("--timeout-ms {text:?}: {error}"))?;
431                self.timeout = Some(millis);
432                Ok(true)
433            }
434            _ => Ok(false),
435        }
436    }
437
438    /// Where the service is and how long to wait, once every flag is read.
439    fn finish(self) -> Result<(Target, Duration), String> {
440        let target = self
441            .target
442            .ok_or("give exactly one of --socket and --config")?;
443        let millis = self.timeout.unwrap_or(DEFAULT_TIMEOUT_MS);
444        Ok((target, Duration::from_millis(millis.get())))
445    }
446}
447
448struct Call {
449    target: Target,
450    timeout: Duration,
451    method: String,
452    /// `None` when the operation has no params.
453    params: Option<Text>,
454}
455
456/// One JSON value as given on the command line, or `-` for standard input.
457enum Text {
458    Given(String),
459    Stdin,
460}
461
462impl Text {
463    fn of(text: String) -> Self {
464        if text == "-" {
465            Self::Stdin
466        } else {
467            Self::Given(text)
468        }
469    }
470}
471
472fn call(args: impl Iterator<Item = String>) -> Result<Reply, Ended> {
473    let asked = parse(args).map_err(|why| Ended::Usage(format!("{why}\n{USAGE}")))?;
474    let params = asked
475        .params
476        .map(|text| one_value(text, "params"))
477        .transpose()?;
478    let socket = socket_of(&asked.target)?;
479    let mut client = Client::connect(&socket, asked.timeout)?;
480    if asked.method == "hello" && params.is_none() {
481        return Ok(Reply::Answered(client.hello().to_owned()));
482    }
483    client.call(&asked.method, params.as_deref())
484}
485
486fn parse(mut args: impl Iterator<Item = String>) -> Result<Call, String> {
487    let mut common = Common::default();
488    let mut rest = Vec::new();
489    while let Some(arg) = args.next() {
490        if common.take(&arg, &mut args)? {
491            continue;
492        }
493        if arg.starts_with("--") {
494            return Err(format!("unknown or repeated flag {arg:?}"));
495        }
496        rest.push(arg);
497    }
498    let (target, timeout) = common.finish()?;
499    let mut rest = rest.into_iter();
500    let method = rest.next().ok_or("name the method to call")?;
501    let params = rest.next().map(Text::of);
502    if let Some(extra) = rest.next() {
503        return Err(format!("unexpected argument {extra:?}"));
504    }
505    Ok(Call {
506        target,
507        timeout,
508        method,
509        params,
510    })
511}
512
513/// `given` as one JSON value, kept as the text given; `what` names it.
514fn one_value(given: Text, what: &str) -> Result<Box<RawValue>, Ended> {
515    let text = match given {
516        Text::Given(text) => text,
517        Text::Stdin => {
518            let mut text = String::new();
519            io::stdin().read_to_string(&mut text).map_err(|error| {
520                Ended::NotAsked(format!(
521                    "the {what} could not be read from standard input: {error}"
522                ))
523            })?;
524            text
525        }
526    };
527    serde_json::from_str(&text)
528        .map_err(|error| Ended::NotAsked(format!("the {what} is not one JSON value: {error}")))
529}
530
531/// The socket named, or the one a service config names, read by the
532/// service's own config parser.
533fn socket_of(target: &Target) -> Result<PathBuf, Ended> {
534    match target {
535        Target::Socket(path) => Ok(path.clone()),
536        Target::Config(path) => {
537            let named = path.display();
538            let bytes = std::fs::read(path).map_err(|error| {
539                Ended::NotAsked(format!(
540                    "the service config {named} could not be read: {error}"
541                ))
542            })?;
543            let config: Config = serde_json::from_slice(&bytes)
544                .map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))?;
545            config
546                .endpoint(false)
547                .map(|endpoint| endpoint.socket)
548                .map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))
549        }
550    }
551}
552
553/// `UnixStream` has no connect timeout, so the connect runs on its own thread
554/// and is waited for at most `timeout`.
555fn connect(socket: &Path, timeout: Duration) -> Result<UnixStream, Ended> {
556    let (sender, receiver) = mpsc::channel();
557    let path = socket.to_path_buf();
558    std::thread::Builder::new()
559        .name("haem-call-connect".to_owned())
560        .spawn(move || {
561            // The receiver is gone only when the connect outlived the timeout.
562            let _unwaited = sender.send(UnixStream::connect(path));
563        })
564        .map_err(|error| Ended::NotAsked(format!("the connect could not start: {error}")))?;
565    match receiver.recv_timeout(timeout) {
566        Ok(Ok(stream)) => Ok(stream),
567        Ok(Err(error)) => Err(Ended::NotAsked(format!(
568            "could not connect to {}: {error}",
569            socket.display()
570        ))),
571        Err(_) => Err(Ended::NotAsked(format!(
572            "could not connect to {} within {} ms",
573            socket.display(),
574            timeout.as_millis()
575        ))),
576    }
577}
578
579#[derive(Serialize)]
580struct Outgoing<'a> {
581    id: &'a str,
582    operation: Operation<'a>,
583}
584
585#[derive(Serialize)]
586struct Operation<'a> {
587    method: &'a str,
588    #[serde(skip_serializing_if = "Option::is_none")]
589    params: Option<&'a RawValue>,
590}
591
592#[derive(Deserialize)]
593struct Incoming {
594    id: String,
595    /// Present-and-null is an answer of null, so `null` is kept as its bytes.
596    #[serde(default, deserialize_with = "present")]
597    result: Option<Box<RawValue>>,
598    #[serde(default)]
599    error: Option<Refusal>,
600}
601
602fn present<'de, D: Deserializer<'de>>(input: D) -> Result<Option<Box<RawValue>>, D::Error> {
603    Box::<RawValue>::deserialize(input).map(Some)
604}
605
606/// The first answer frame read only for whether it opens a sealed receipt.
607#[derive(Deserialize)]
608struct Probe {
609    #[serde(default)]
610    receipt: Option<Box<RawValue>>,
611}
612
613fn exchange(
614    stream: &mut UnixStream,
615    limit: NonZeroU32,
616    method: &str,
617    params: Option<&RawValue>,
618) -> Result<Reply, Ended> {
619    let operation = Operation { method, params };
620    let request = serde_json::to_vec(&Outgoing { id: ID, operation })
621        .map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))?;
622    if !u32::try_from(request.len()).is_ok_and(|length| length <= limit.get()) {
623        return Err(Ended::NotAsked(format!(
624            "request_too_large: the request is {} bytes and the service reads at most {limit}",
625            request.len()
626        )));
627    }
628    frame::write(stream, &request, limit).map_err(|error| lost("sending the request", &error))?;
629    receive(stream, limit)
630}
631
632/// Read one answer frame, and the rest of a sealed receipt it opens.
633fn receive(stream: &mut UnixStream, limit: NonZeroU32) -> Result<Reply, Ended> {
634    let bytes = match frame::read(stream, limit) {
635        Ok(Some(bytes)) => bytes,
636        Ok(None) => {
637            return Err(Ended::Unknown(
638                "the service closed the connection before answering".to_owned(),
639            ));
640        }
641        Err(error) => return Err(lost("waiting for the answer", &error)),
642    };
643    let probe: Probe = serde_json::from_slice(&bytes)
644        .map_err(|error| Ended::Unknown(format!("receipt_frame_decode_failed: {error}")))?;
645    if probe.receipt.is_none() {
646        return reply(&bytes);
647    }
648    let mut assembler =
649        Assembler::new(ID, limit.get()).map_err(|error| Ended::Unknown(error.to_string()))?;
650    let mut body = bytes;
651    loop {
652        if let Some(complete) = assembler
653            .accept(&body)
654            .map_err(|error| Ended::Unknown(error.to_string()))?
655        {
656            let complete = Arc::new(complete);
657            let refusal = complete.refusal().ok_or_else(|| Ended::Received {
658                why: "receipt_refusal_missing".to_owned(),
659                receipt: Arc::clone(&complete),
660            })?;
661            return Ok(Reply::Refused(Refusal {
662                code: refusal.code.clone(),
663                message: refusal.message.clone(),
664                path: refusal.path.clone(),
665                verified: Some(complete),
666            }));
667        }
668        body = match frame::read(stream, limit) {
669            Ok(Some(body)) => body,
670            Ok(None) => return Err(Ended::Unknown("receipt_connection_closed".to_owned())),
671            Err(error) => {
672                return Err(Ended::Unknown(format!(
673                    "receipt_connection_failed: {error}"
674                )));
675            }
676        };
677    }
678}
679
680fn lost(doing: &str, error: &io::Error) -> Ended {
681    match error.kind() {
682        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => {
683            Ended::Unknown(format!("timed out {doing}: {error}"))
684        }
685        _ => Ended::Unknown(format!("the connection failed {doing}: {error}")),
686    }
687}
688
689fn reply(bytes: &[u8]) -> Result<Reply, Ended> {
690    let unreadable = |why: String| Ended::Unknown(format!("the answer could not be read: {why}"));
691    let incoming: Incoming =
692        serde_json::from_slice(bytes).map_err(|error| unreadable(error.to_string()))?;
693    if incoming.id != ID {
694        return Err(unreadable(format!(
695            "it answers {:?}, not {ID:?}",
696            incoming.id
697        )));
698    }
699    match (incoming.result, incoming.error) {
700        (Some(result), None) => Ok(Reply::Answered(result)),
701        (None, Some(refusal)) => Ok(Reply::Refused(refusal)),
702        _ => Err(unreadable(
703            "it carries neither or both of result and error".to_owned(),
704        )),
705    }
706}
707
708#[cfg(test)]
709mod nonzero_tests {
710    use super::{Common, DEFAULT_TIMEOUT_MS, NonZeroU64, PathBuf, Target};
711
712    #[test]
713    fn nonzero_timeout_readback() -> Result<(), Box<dyn std::error::Error>> {
714        let cases = [
715            (None, DEFAULT_TIMEOUT_MS.get()),
716            (Some(1), 1),
717            (Some(37), 37),
718            (Some(u64::MAX), u64::MAX),
719        ];
720        for (given, expected) in cases {
721            let timeout = given
722                .map(|value| NonZeroU64::new(value).ok_or("zero timeout"))
723                .transpose()?;
724            let common = Common {
725                target: Some(Target::Socket(PathBuf::from("unused.sock"))),
726                timeout: std::hint::black_box(timeout),
727            };
728            let (_, actual) = common.finish()?;
729            assert_eq!(actual.as_millis(), u128::from(expected));
730        }
731        Ok(())
732    }
733}
734
735#[cfg(test)]
736mod json_text_refusal_tests {
737    use super::{KNOWN_REFUSALS, Refusal, Reply, finish, refused};
738    use std::process::ExitCode;
739
740    #[test]
741    fn json_text_invalid_is_a_named_client_refusal() {
742        let refusal = Refusal {
743            code: "json_text_invalid".to_owned(),
744            message: "value is not JSON".to_owned(),
745            path: Some("/payload".to_owned()),
746            verified: None,
747        };
748        assert!(KNOWN_REFUSALS.contains(&refusal.code.as_str()));
749        assert_eq!(
750            refused(&refusal),
751            "haematite: refused json_text_invalid: value is not JSON at /payload"
752        );
753        assert_eq!(finish(Ok(Reply::Refused(refusal))), ExitCode::from(2));
754    }
755}
756
757#[cfg(test)]
758mod access_refusal_tests {
759    use super::KNOWN_REFUSALS;
760
761    /// A change a Lys-served service refused before it ran is a refusal, not
762    /// an unknown outcome: the client says it was not applied.
763    #[test]
764    fn every_access_refusal_is_known_not_to_have_applied() {
765        for code in [
766            "route_unclassified",
767            "permission_refused",
768            "held_for_approval",
769            "pass_missing",
770            "lys_could_not_be_asked",
771            "signature_refused",
772            "pass_expired",
773        ] {
774            assert!(KNOWN_REFUSALS.contains(&code), "{code} is not known");
775        }
776    }
777}