haematite_cli/

client.rs

//! `haematite call`: ask a running service anything its protocol offers, from the
//! command line; `haematite write` is in [`write`] and `haematite query` in [`query`].
//!
//! The client says hello first and takes the service's `max_frame` as the
//! bound on what it sends and what it reads. Params go out as the text given
//! and a result comes back as the bytes the service sent: no number is parsed
//! into a float and written again. Four outcomes are told apart by exit code:
//! answered (0), could not ask, so nothing was sent (1), refused by the service
//! (2), and outcome not known to the caller (3).

use crate::serve::protocol::{HelloAgreement, PROTOCOL, RECEIPT_CAPABILITY};
use crate::serve::receipt_fragments::{Assembler, VerifiedReceipt, minimum_cap};
use std::io::{self, Read, Write};
use std::num::{NonZeroU32, NonZeroU64};
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::{Arc, mpsc};
use std::time::Duration;

use serde::{Deserialize, Deserializer, Serialize};
use serde_json::value::RawValue;

use crate::serve::{cli::Config, frame};

pub mod network;
pub mod query;
pub mod write;

pub use network::Address;
use network::{Hangup, Stream};

const USAGE: &str = "\
usage: haematite call (--socket <path> | --config <serve.json> | --endpoint <host:port> --server-name <name>) [--pass-file <path>] [--timeout-ms N] <method> [<params> | -]
<params> is one JSON value; - reads it from standard input; absent, the operation has no params.
--timeout-ms bounds connecting, writing and reading; the default is 30000.
Exit 0: answered, the result on standard output. Exit 1: could not ask; nothing was sent.
Exit 2: refused by the service, its code and message on standard error. Exit 3: the outcome
is not known to the caller; the service may have acted.";

const DEFAULT_TIMEOUT_MS: NonZeroU64 = NonZeroU64::MIN.saturating_add(29_999);
/// Hello's answer is read before the service has said its frame limit.
const HELLO_BOUND: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
const ID: &str = "haem-call";

/// Codes which leave the mutation's outcome uncertain.
///
/// Unrecognised codes also retain uncertainty unless the shared refusal table
/// proves otherwise.
pub const MAY_HAVE_APPLIED: &[&str] = &[
    // The commit failed after the branch reference moved: storage/mutations.rs
    // 196 and 476 (a fork); a discard, storage/discards.rs 226, 296, 332, 418
    // and 424; a receipt's confirmation, storage/mutations.rs 71.
    "durability_unknown",
    // The change is durable and the catalogue could not show it:
    // storage/catalogue.rs 37, by publish_record after every commit.
    "committed_unpublished",
    // The reference vanished after its commit: storage.rs 275, the same
    // publish_record; it is also raised before commits, which cannot be told
    // apart from here.
    "reference_missing",
    // A refused discard commit whose intent could not be withdrawn:
    // storage/discards.rs 253.
    "discard_abort_failed",
    // The catalogue could not hide a branch whose discard intent was
    // installed: storage/catalogue.rs 49, called from storage/discards.rs 221
    // and after the commit at 416, 425 and 429.
    "catalogue_unavailable",
    // Raised by Failure::storage everywhere, including after the change, as
    // when the receipt cannot be serialized (serve/service.rs 123-124).
    "storage_unavailable",
    // The answer did not fit after the change was made: serve/service.rs 108
    // and serve/transport.rs 177, which says to reconcile by request id.
    "response_limit",
];

/// The service's shared proof that a refusal applied no mutation.
pub use crate::serve::protocol::NOT_APPLIED_REFUSALS as KNOWN_REFUSALS;

/// Why an exchange gave neither an answer nor a refusal.
#[derive(Debug)]
pub enum Ended {
    /// The command line was not one the command takes; the usage text follows.
    Usage(String),
    /// Nothing the service could act on was sent.
    NotAsked(String),
    /// The request was sent, or may have been; whether the service acted on
    /// it is not known.
    Unknown(String),
    /// The service sent a complete, verified receipt, and whether the change
    /// was applied is still not known: the receipt carried no refusal, or its
    /// code is one that may have applied or that this client does not know.
    /// The receipt is kept as the exact bytes sent, to reconcile by.
    Received {
        why: String,
        receipt: Arc<VerifiedReceipt>,
    },
}

impl Ended {
    /// The cause, without the words for which kind of end it was.
    #[must_use]
    pub fn why(&self) -> &str {
        match self {
            Self::Usage(why)
            | Self::NotAsked(why)
            | Self::Unknown(why)
            | Self::Received { why, .. } => why,
        }
    }
}

impl std::fmt::Display for Ended {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let kind = match self {
            Self::Usage(_) => "usage",
            Self::NotAsked(_) => "not asked",
            Self::Unknown(_) => "outcome unknown",
            Self::Received { .. } => "complete receipt received",
        };
        write!(formatter, "{kind}: {}", self.why())
    }
}

impl std::error::Error for Ended {}

/// What the service said to one request.
#[derive(Debug)]
pub enum Reply {
    /// The bytes the service sent for `result`.
    Answered(Box<RawValue>),
    Refused(Refusal),
}

/// A refusal by name, as the service sent it.
#[derive(Debug, Deserialize)]
pub struct Refusal {
    pub code: String,
    pub message: String,
    #[serde(default)]
    pub path: Option<String>,
    #[serde(skip)]
    pub verified: Option<Arc<VerifiedReceipt>>,
}

/// One connection to a running service, past its hello.
pub struct Client {
    stream: Stream,
    hello: Box<RawValue>,
    max_frame: NonZeroU32,
    max_page: NonZeroU32,
}

#[derive(Deserialize)]
struct Limits {
    protocol: String,
    capabilities: Vec<String>,
    max_frame: NonZeroU32,
    max_page: NonZeroU32,
}

impl Client {
    /// Connect to `socket` and say hello, each step bounded by `timeout`.
    ///
    /// # Errors
    ///
    /// `Ended::NotAsked` naming the cause: hello changes nothing, so any
    /// failure up to its answer leaves nothing asked.
    pub fn connect(socket: &Path, timeout: Duration) -> Result<Self, Ended> {
        Self::over(Stream::Unix(connect(socket, timeout)?), timeout)
    }

    /// Reach the store at `address`, its socket or its network endpoint
    /// over verified TLS (HAEMNET-003 R1), and say hello, each step bounded
    /// by `timeout`.
    ///
    /// # Errors
    ///
    /// `Ended::NotAsked` naming the cause, as [`Client::connect`]: a network
    /// store not reached, or not over verified TLS, was asked nothing.
    pub fn reach(address: &Address, timeout: Duration) -> Result<Self, Ended> {
        match address {
            Address::Socket(socket) => Self::connect(socket, timeout),
            Address::Network(remote) => Self::over(
                network::open(remote, timeout)
                    .map_err(|refused| Ended::NotAsked(refused.why().to_owned()))?,
                timeout,
            ),
        }
    }

    /// Say hello over `stream`, each step bounded by `timeout`.
    fn over(mut stream: Stream, timeout: Duration) -> Result<Self, Ended> {
        let unset =
            |error: io::Error| Ended::NotAsked(format!("the timeout could not be set: {error}"));
        stream.set_read_timeout(Some(timeout)).map_err(unset)?;
        stream.set_write_timeout(Some(timeout)).map_err(unset)?;
        let agreement = serde_json::value::to_raw_value(&HelloAgreement::client())
            .map_err(|error| Ended::NotAsked(format!("receipt_hello_encode_failed: {error}")))?;
        let hello = match exchange(&mut stream, HELLO_BOUND, "hello", Some(&agreement)) {
            Ok(Reply::Answered(hello)) => hello,
            Ok(Reply::Refused(refusal)) => {
                return Err(Ended::NotAsked(format!(
                    "the service refused hello: {}: {}",
                    refusal.code, refusal.message
                )));
            }
            Err(Ended::Received { why, receipt }) => {
                return Err(Ended::Received {
                    why: format!("hello failed: {why}"),
                    receipt,
                });
            }
            Err(Ended::Unknown(why) | Ended::NotAsked(why) | Ended::Usage(why)) => {
                return Err(Ended::NotAsked(format!("hello failed: {why}")));
            }
        };
        let limits: Limits = serde_json::from_str(hello.get()).map_err(|error| {
            Ended::NotAsked(format!(
                "the service's hello did not give its limits: {error}"
            ))
        })?;
        if limits.protocol != PROTOCOL {
            return Err(Ended::NotAsked(format!(
                "receipt_protocol_mismatch: expected {PROTOCOL}, received {}",
                limits.protocol
            )));
        }
        if !limits
            .capabilities
            .iter()
            .any(|capability| capability == RECEIPT_CAPABILITY)
        {
            return Err(Ended::NotAsked(format!(
                "typed_receipt_capability_missing: required {RECEIPT_CAPABILITY}"
            )));
        }
        let minimum = minimum_cap().map_err(|error| Ended::NotAsked(error.to_string()))?;
        if u64::from(limits.max_frame.get())
            < u64::try_from(minimum).map_err(|error| Ended::NotAsked(error.to_string()))?
        {
            return Err(Ended::NotAsked(format!(
                "receipt_frame_too_small: offered {}, minimum {minimum}",
                limits.max_frame
            )));
        }
        Ok(Self {
            stream,
            hello,
            max_frame: limits.max_frame,
            max_page: limits.max_page,
        })
    }

    /// The bytes the service sent for hello's `result`.
    #[must_use]
    pub fn hello(&self) -> &RawValue {
        &self.hello
    }

    /// The largest frame the service reads or sends.
    #[must_use]
    pub const fn max_frame(&self) -> NonZeroU32 {
        self.max_frame
    }

    /// The most rows the service puts on one page.
    #[must_use]
    pub const fn max_page(&self) -> NonZeroU32 {
        self.max_page
    }

    /// Send one request and read its reply.
    ///
    /// # Errors
    ///
    /// `Ended::NotAsked` for a request larger than the service reads, refused
    /// before sending; `Ended::Unknown` for any failure from sending onward,
    /// and for a `mutation.apply` error that is not one of
    /// [`KNOWN_REFUSALS`]: those the service sends when the change may already
    /// have been applied ([`MAY_HAVE_APPLIED`]), and any code this client does
    /// not know.
    pub fn call(&mut self, method: &str, params: Option<&RawValue>) -> Result<Reply, Ended> {
        let reply = exchange(&mut self.stream, self.max_frame, method, params)?;
        let Reply::Refused(refusal) = reply else {
            return Ok(reply);
        };
        // A read changes nothing, so a refusal of a read is a refusal.
        if method != "mutation.apply" || KNOWN_REFUSALS.contains(&refusal.code.as_str()) {
            return Ok(Reply::Refused(refusal));
        }
        let why = if MAY_HAVE_APPLIED.contains(&refusal.code.as_str()) {
            "the change may have been applied"
        } else {
            "this client does not know that code, so whether the change was applied is not known"
        };
        let why = format!(
            "the service answered {}: {}; {why}",
            refusal.code, refusal.message
        );
        match refusal.verified {
            Some(receipt) => Err(Ended::Received { why, receipt }),
            None => Err(Ended::Unknown(why)),
        }
    }

    /// Read the next frame of an answer that is a stream, such as
    /// `head.follow`'s, sent under the same id as its first. A read changes
    /// nothing, so a refusal is a refusal.
    ///
    /// # Errors
    ///
    /// `Ended::Unknown` when the frame could not be read: the connection
    /// closed, failed or timed out.
    pub fn next_reply(&mut self) -> Result<Reply, Ended> {
        receive(&mut self.stream, self.max_frame)
    }

    /// An owned handle used only to cancel a disconnected browser's follow.
    pub(crate) fn follow_socket(&self) -> io::Result<Hangup> {
        self.stream.hangup()
    }

    /// Bound each later read by `timeout`, or by nothing with `None`.
    ///
    /// # Errors
    ///
    /// `Ended::Unknown` when the bound could not be set.
    pub fn read_within(&self, timeout: Option<Duration>) -> Result<(), Ended> {
        self.stream
            .set_read_timeout(timeout)
            .map_err(|error| Ended::Unknown(format!("the read bound could not be set: {error}")))
    }
}

/// What asking a network store hello came to (HAEMNET-003 R1).
pub enum Reached {
    /// Reached over verified TLS, and the store answered hello.
    Answered,
    /// Reached over verified TLS; hello was not answered, for the reason.
    Unanswered(String),
    /// Not reached over verified TLS.
    Refused(network::Refused),
}

/// Reach `remote` over verified TLS and ask it hello, which changes
/// nothing, each step bounded by `timeout`.
#[must_use]
pub fn probe(remote: &network::Remote, timeout: Duration) -> Reached {
    match network::open(remote, timeout) {
        Err(refused) => Reached::Refused(refused),
        Ok(stream) => match Client::over(stream, timeout) {
            Ok(_) => Reached::Answered,
            Err(ended) => Reached::Unanswered(ended.why().to_owned()),
        },
    }
}

/// `haematite call`, ending in its exit code.
#[must_use]
pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
    finish(call(args))
}

/// Print what an exchange came to and give its exit code: 0 answered, 1 could
/// not ask, 2 refused, 3 outcome not known to the caller.
fn finish(outcome: Result<Reply, Ended>) -> ExitCode {
    match outcome {
        Ok(Reply::Answered(result)) => {
            let mut output = io::stdout().lock();
            match writeln!(output, "{}", result.get()).and_then(|()| output.flush()) {
                Ok(()) => ExitCode::SUCCESS,
                Err(error) => {
                    // The service acted and the caller did not get its answer.
                    eprintln!(
                        "haematite: answer received, but the result could not be written to stdout: {error}"
                    );
                    ExitCode::from(3)
                }
            }
        }
        Ok(Reply::Refused(refusal)) => {
            if let Err(error) = print_refusal(&refusal) {
                eprintln!("haematite: receipt_output_failed: {error}");
            }
            ExitCode::from(2)
        }
        Err(ended) => {
            if let Err(error) = print_ended(&ended) {
                eprintln!("haematite: receipt_output_failed: {error}");
            }
            ExitCode::from(exit_code(&ended))
        }
    }
}

/// 1 when nothing was sent, so asking again is safe; 3 when it may have been.
#[must_use]
pub const fn exit_code(ended: &Ended) -> u8 {
    match ended {
        Ended::Usage(_) | Ended::NotAsked(_) => 1,
        Ended::Unknown(_) | Ended::Received { .. } => 3,
    }
}

/// The standard error line for a refusal.
#[must_use]
pub fn refused(refusal: &Refusal) -> String {
    let at = refusal
        .path
        .as_deref()
        .map_or_else(String::new, |path| format!(" at {path}"));
    let (code, message) = (&refusal.code, &refusal.message);
    format!("haematite: refused {code}: {message}{at}")
}

/// The standard error line for an exchange that gave no reply.
#[must_use]
pub fn words(ended: &Ended) -> String {
    match ended {
        Ended::Usage(why) => format!("haematite: {why}"),
        Ended::NotAsked(why) => format!("haematite: could not ask: {why}"),
        Ended::Unknown(why) => format!("haematite: the outcome is not known: {why}"),
        Ended::Received { why, .. } => format!("haematite: complete receipt received: {why}"),
    }
}

pub(super) fn print_refusal(refusal: &Refusal) -> io::Result<()> {
    let mut output = io::stderr().lock();
    writeln!(output, "{}", refused(refusal))?;
    if let Some(receipt) = &refusal.verified {
        output.write_all(receipt.bytes())?;
        output.write_all(b"\n")?;
    }
    output.flush()
}
pub(super) fn print_ended(ended: &Ended) -> io::Result<()> {
    let mut output = io::stderr().lock();
    writeln!(output, "{}", words(ended))?;
    if let Ended::Received { receipt, .. } = ended {
        output.write_all(receipt.bytes())?;
        output.write_all(b"\n")?;
    }
    output.flush()
}

enum Place {
    Socket(PathBuf),
    Config(PathBuf),
    /// A store's HAEMNET-001 network endpoint, over TLS verified against
    /// this machine's roots, as the name its certificate carries
    /// (HAEMNET-004 R1).
    Endpoint {
        endpoint: String,
        server_name: String,
    },
}

/// Where a client command's store is, and the file holding the pass it
/// presents there: required on the network, which admits only a Lys pass
/// (HAEMNET-001 R2), and presented on the socket when given.
struct Target {
    place: Place,
    pass: Option<PathBuf>,
}

impl Target {
    /// Reach the store and say hello, then present the pass, when there is
    /// one, with `session.pass`.
    fn open(&self, timeout: Duration) -> Result<Client, Ended> {
        let address = match &self.place {
            Place::Socket(socket) => network::Address::Socket(socket.clone()),
            Place::Config(_) => network::Address::Socket(socket_of(&self.place)?),
            Place::Endpoint {
                endpoint,
                server_name,
            } => {
                let trust = network::Trust::system().map_err(Ended::NotAsked)?;
                network::Address::Network(
                    network::Remote::new(endpoint, server_name, &trust).map_err(Ended::NotAsked)?,
                )
            }
        };
        let mut client = Client::reach(&address, timeout)?;
        if let Some(file) = &self.pass {
            let named = file.display();
            let pass = std::fs::read_to_string(file).map_err(|error| {
                Ended::NotAsked(format!("--pass-file {named}: {}", error.kind()))
            })?;
            let pass = pass.trim();
            if pass.is_empty() {
                return Err(Ended::NotAsked(format!("--pass-file {named}: empty")));
            }
            let params = serde_json::value::to_raw_value(&serde_json::json!({ "pass": pass }))
                .map_err(|_| Ended::NotAsked("the pass could not be written".to_owned()))?;
            if let Reply::Refused(refusal) = client.call("session.pass", Some(&params))? {
                return Err(Ended::NotAsked(format!(
                    "the store refused the pass: {}: {}",
                    refusal.code, refusal.message
                )));
            }
        }
        Ok(client)
    }
}

const ONE_PLACE: &str = "give exactly one of --socket, --config and --endpoint";

/// The flags every client command takes: where the service is, the pass it
/// is given, and how long to wait for it.
#[derive(Default)]
struct Common {
    place: Option<Place>,
    endpoint: Option<String>,
    server_name: Option<String>,
    pass: Option<PathBuf>,
    timeout: Option<NonZeroU64>,
}

impl Common {
    /// Take `arg`, and its value from `args`, when it is a shared flag.
    fn take(&mut self, arg: &str, args: &mut impl Iterator<Item = String>) -> Result<bool, String> {
        match arg {
            "--socket" | "--config" if self.place.is_none() => {
                let path = PathBuf::from(args.next().ok_or_else(|| format!("{arg} needs a path"))?);
                self.place = Some(if arg == "--socket" {
                    Place::Socket(path)
                } else {
                    Place::Config(path)
                });
                Ok(true)
            }
            "--endpoint" if self.endpoint.is_none() => {
                self.endpoint = Some(args.next().ok_or("--endpoint needs <host>:<port>")?);
                Ok(true)
            }
            "--server-name" if self.server_name.is_none() => {
                self.server_name = Some(args.next().ok_or("--server-name needs a name")?);
                Ok(true)
            }
            "--pass-file" if self.pass.is_none() => {
                self.pass = Some(PathBuf::from(
                    args.next().ok_or("--pass-file needs a path")?,
                ));
                Ok(true)
            }
            "--socket" | "--config" => Err(ONE_PLACE.to_owned()),
            "--timeout-ms" if self.timeout.is_none() => {
                let text = args
                    .next()
                    .ok_or("--timeout-ms needs a number of milliseconds")?;
                let millis = text
                    .parse::<NonZeroU64>()
                    .map_err(|error| format!("--timeout-ms {text:?}: {error}"))?;
                self.timeout = Some(millis);
                Ok(true)
            }
            _ => Ok(false),
        }
    }

    /// Where the service is and how long to wait, once every flag is read.
    fn finish(self) -> Result<(Target, Duration), String> {
        let place = match (self.place, self.endpoint, self.server_name) {
            (Some(place), None, None) => place,
            (None, Some(endpoint), Some(server_name)) => {
                if self.pass.is_none() {
                    return Err(
                        "--endpoint needs --pass-file: the network admits only a pass".to_owned(),
                    );
                }
                Place::Endpoint {
                    endpoint,
                    server_name,
                }
            }
            (None, Some(_), None) => return Err("--endpoint needs --server-name".to_owned()),
            (_, None, Some(_)) => {
                return Err("--server-name is given only with --endpoint".to_owned());
            }
            (None, None, None) | (Some(_), Some(_), _) => return Err(ONE_PLACE.to_owned()),
        };
        let millis = self.timeout.unwrap_or(DEFAULT_TIMEOUT_MS);
        Ok((
            Target {
                place,
                pass: self.pass,
            },
            Duration::from_millis(millis.get()),
        ))
    }
}

struct Call {
    target: Target,
    timeout: Duration,
    method: String,
    /// `None` when the operation has no params.
    params: Option<Text>,
}

/// One JSON value as given on the command line, or `-` for standard input.
enum Text {
    Given(String),
    Stdin,
}

impl Text {
    fn of(text: String) -> Self {
        if text == "-" {
            Self::Stdin
        } else {
            Self::Given(text)
        }
    }
}

fn call(args: impl Iterator<Item = String>) -> Result<Reply, Ended> {
    let asked = parse(args).map_err(|why| Ended::Usage(format!("{why}\n{USAGE}")))?;
    let params = asked
        .params
        .map(|text| one_value(text, "params"))
        .transpose()?;
    let mut client = asked.target.open(asked.timeout)?;
    if asked.method == "hello" && params.is_none() {
        return Ok(Reply::Answered(client.hello().to_owned()));
    }
    client.call(&asked.method, params.as_deref())
}

fn parse(mut args: impl Iterator<Item = String>) -> Result<Call, String> {
    let mut common = Common::default();
    let mut rest = Vec::new();
    while let Some(arg) = args.next() {
        if common.take(&arg, &mut args)? {
            continue;
        }
        if arg.starts_with("--") {
            return Err(format!("unknown or repeated flag {arg:?}"));
        }
        rest.push(arg);
    }
    let (target, timeout) = common.finish()?;
    let mut rest = rest.into_iter();
    let method = rest.next().ok_or("name the method to call")?;
    let params = rest.next().map(Text::of);
    if let Some(extra) = rest.next() {
        return Err(format!("unexpected argument {extra:?}"));
    }
    Ok(Call {
        target,
        timeout,
        method,
        params,
    })
}

/// `given` as one JSON value, kept as the text given; `what` names it.
fn one_value(given: Text, what: &str) -> Result<Box<RawValue>, Ended> {
    let text = match given {
        Text::Given(text) => text,
        Text::Stdin => {
            let mut text = String::new();
            io::stdin().read_to_string(&mut text).map_err(|error| {
                Ended::NotAsked(format!(
                    "the {what} could not be read from standard input: {error}"
                ))
            })?;
            text
        }
    };
    serde_json::from_str(&text)
        .map_err(|error| Ended::NotAsked(format!("the {what} is not one JSON value: {error}")))
}

/// The socket named, or the one a service config names, read by the
/// service's own config parser.
fn socket_of(place: &Place) -> Result<PathBuf, Ended> {
    match place {
        Place::Socket(path) => Ok(path.clone()),
        Place::Endpoint { endpoint, .. } => Err(Ended::NotAsked(format!(
            "{endpoint} is a network endpoint, not a socket"
        ))),
        Place::Config(path) => {
            let named = path.display();
            let bytes = std::fs::read(path).map_err(|error| {
                Ended::NotAsked(format!(
                    "the service config {named} could not be read: {error}"
                ))
            })?;
            let config: Config = serde_json::from_slice(&bytes)
                .map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))?;
            config
                .endpoint(false)
                .map(|endpoint| endpoint.socket)
                .map_err(|error| Ended::NotAsked(format!("the service config {named}: {error}")))
        }
    }
}

/// `UnixStream` has no connect timeout, so the connect runs on its own thread
/// and is waited for at most `timeout`.
fn connect(socket: &Path, timeout: Duration) -> Result<UnixStream, Ended> {
    let (sender, receiver) = mpsc::channel();
    let path = socket.to_path_buf();
    std::thread::Builder::new()
        .name("haem-call-connect".to_owned())
        .spawn(move || {
            // The receiver is gone only when the connect outlived the timeout.
            let _unwaited = sender.send(UnixStream::connect(path));
        })
        .map_err(|error| Ended::NotAsked(format!("the connect could not start: {error}")))?;
    match receiver.recv_timeout(timeout) {
        Ok(Ok(stream)) => Ok(stream),
        Ok(Err(error)) => Err(Ended::NotAsked(format!(
            "could not connect to {}: {error}",
            socket.display()
        ))),
        Err(_) => Err(Ended::NotAsked(format!(
            "could not connect to {} within {} ms",
            socket.display(),
            timeout.as_millis()
        ))),
    }
}

#[derive(Serialize)]
struct Outgoing<'a> {
    id: &'a str,
    operation: Operation<'a>,
}

#[derive(Serialize)]
struct Operation<'a> {
    method: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    params: Option<&'a RawValue>,
}

#[derive(Deserialize)]
struct Incoming {
    id: String,
    /// Present-and-null is an answer of null, so `null` is kept as its bytes.
    #[serde(default, deserialize_with = "present")]
    result: Option<Box<RawValue>>,
    #[serde(default)]
    error: Option<Refusal>,
}

fn present<'de, D: Deserializer<'de>>(input: D) -> Result<Option<Box<RawValue>>, D::Error> {
    Box::<RawValue>::deserialize(input).map(Some)
}

/// The first answer frame read only for whether it opens a sealed receipt.
#[derive(Deserialize)]
struct Probe {
    #[serde(default)]
    receipt: Option<Box<RawValue>>,
}

fn exchange(
    stream: &mut (impl Read + Write),
    limit: NonZeroU32,
    method: &str,
    params: Option<&RawValue>,
) -> Result<Reply, Ended> {
    let operation = Operation { method, params };
    let request = serde_json::to_vec(&Outgoing { id: ID, operation })
        .map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))?;
    if !u32::try_from(request.len()).is_ok_and(|length| length <= limit.get()) {
        return Err(Ended::NotAsked(format!(
            "request_too_large: the request is {} bytes and the service reads at most {limit}",
            request.len()
        )));
    }
    frame::write(stream, &request, limit).map_err(|error| lost("sending the request", &error))?;
    receive(stream, limit)
}

/// Read one answer frame, and the rest of a sealed receipt it opens.
fn receive(stream: &mut impl Read, limit: NonZeroU32) -> Result<Reply, Ended> {
    let bytes = match frame::read(stream, limit) {
        Ok(Some(bytes)) => bytes,
        Ok(None) => {
            return Err(Ended::Unknown(
                "the service closed the connection before answering".to_owned(),
            ));
        }
        Err(error) => return Err(lost("waiting for the answer", &error)),
    };
    let probe: Probe = serde_json::from_slice(&bytes)
        .map_err(|error| Ended::Unknown(format!("receipt_frame_decode_failed: {error}")))?;
    if probe.receipt.is_none() {
        return reply(&bytes);
    }
    let mut assembler =
        Assembler::new(ID, limit.get()).map_err(|error| Ended::Unknown(error.to_string()))?;
    let mut body = bytes;
    loop {
        if let Some(complete) = assembler
            .accept(&body)
            .map_err(|error| Ended::Unknown(error.to_string()))?
        {
            let complete = Arc::new(complete);
            let refusal = complete.refusal().ok_or_else(|| Ended::Received {
                why: "receipt_refusal_missing".to_owned(),
                receipt: Arc::clone(&complete),
            })?;
            return Ok(Reply::Refused(Refusal {
                code: refusal.code.clone(),
                message: refusal.message.clone(),
                path: refusal.path.clone(),
                verified: Some(complete),
            }));
        }
        body = match frame::read(stream, limit) {
            Ok(Some(body)) => body,
            Ok(None) => return Err(Ended::Unknown("receipt_connection_closed".to_owned())),
            Err(error) => {
                return Err(Ended::Unknown(format!(
                    "receipt_connection_failed: {error}"
                )));
            }
        };
    }
}

fn lost(doing: &str, error: &io::Error) -> Ended {
    match error.kind() {
        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => {
            Ended::Unknown(format!("timed out {doing}: {error}"))
        }
        _ => Ended::Unknown(format!("the connection failed {doing}: {error}")),
    }
}

fn reply(bytes: &[u8]) -> Result<Reply, Ended> {
    let unreadable = |why: String| Ended::Unknown(format!("the answer could not be read: {why}"));
    let incoming: Incoming =
        serde_json::from_slice(bytes).map_err(|error| unreadable(error.to_string()))?;
    if incoming.id != ID {
        return Err(unreadable(format!(
            "it answers {:?}, not {ID:?}",
            incoming.id
        )));
    }
    match (incoming.result, incoming.error) {
        (Some(result), None) => Ok(Reply::Answered(result)),
        (None, Some(refusal)) => Ok(Reply::Refused(refusal)),
        _ => Err(unreadable(
            "it carries neither or both of result and error".to_owned(),
        )),
    }
}

#[cfg(test)]
mod nonzero_tests {
    use super::{Common, DEFAULT_TIMEOUT_MS, NonZeroU64, PathBuf, Place};

    #[test]
    fn nonzero_timeout_readback() -> Result<(), Box<dyn std::error::Error>> {
        let cases = [
            (None, DEFAULT_TIMEOUT_MS.get()),
            (Some(1), 1),
            (Some(37), 37),
            (Some(u64::MAX), u64::MAX),
        ];
        for (given, expected) in cases {
            let timeout = given
                .map(|value| NonZeroU64::new(value).ok_or("zero timeout"))
                .transpose()?;
            let common = Common {
                place: Some(Place::Socket(PathBuf::from("unused.sock"))),
                timeout: std::hint::black_box(timeout),
                ..Common::default()
            };
            let (_, actual) = common.finish()?;
            assert_eq!(actual.as_millis(), u128::from(expected));
        }
        Ok(())
    }
}

#[cfg(test)]
mod json_text_refusal_tests {
    use super::{KNOWN_REFUSALS, Refusal, Reply, finish, refused};
    use std::process::ExitCode;

    #[test]
    fn json_text_invalid_is_a_named_client_refusal() {
        let refusal = Refusal {
            code: "json_text_invalid".to_owned(),
            message: "value is not JSON".to_owned(),
            path: Some("/payload".to_owned()),
            verified: None,
        };
        assert!(KNOWN_REFUSALS.contains(&refusal.code.as_str()));
        assert_eq!(
            refused(&refusal),
            "haematite: refused json_text_invalid: value is not JSON at /payload"
        );
        assert_eq!(finish(Ok(Reply::Refused(refusal))), ExitCode::from(2));
    }
}

#[cfg(test)]
mod access_refusal_tests {
    use super::KNOWN_REFUSALS;

    /// A change a Lys-served service refused before it ran is a refusal, not
    /// an unknown outcome: the client says it was not applied.
    #[test]
    fn every_access_refusal_is_known_not_to_have_applied() {
        for code in [
            "route_unclassified",
            "permission_refused",
            "held_for_approval",
            "pass_missing",
            "lys_could_not_be_asked",
            "signature_refused",
            "pass_expired",
        ] {
            assert!(KNOWN_REFUSALS.contains(&code), "{code} is not known");
        }
    }
}