haematite-cli 0.21.7

//! Manual local Unix-socket service with bounded clients and frame deadlines.

use std::io::{self, Read, Write};
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::net::UnixStream;
use std::time::{Duration, Instant};

use serde_json::json;

use super::{
    frame,
    protocol::{
        DecodedRequest, MutationContext, Operation, Request, ResponseFailure, decode_request,
    },
    service::Service,
    storage::Failure,
};

#[path = "transport_clients.rs"]
mod clients;
#[path = "transport_listener.rs"]
mod listener;
#[cfg(test)]
use clients::{Clients, Permit};
pub use listener::Listener;

#[derive(Clone, Copy, Debug)]
pub struct Limits {
    pub max_frame: NonZeroU32,
    pub max_clients: NonZeroUsize,
    pub frame_timeout: Duration,
}

/// The log line for a connection that ended in an error, saying which kind of
/// end it was. A client that closes between frames ends with no error at all.
fn ending(error: &io::Error) -> String {
    let what = match error.kind() {
        io::ErrorKind::UnexpectedEof => "closed inside a frame",
        // an idle held connection ends here too; macOS names it EAGAIN
        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => {
            "sent no whole frame within the frame deadline"
        }
        io::ErrorKind::InvalidData => "sent a frame that was refused",
        io::ErrorKind::BrokenPipe => "closed before its answer was written",
        _ => "connection failed",
    };
    format!("haematite client {what}: {error}")
}

#[derive(Debug)]
struct ReceiptDeliveryError {
    original: ResponseFailure,
    sealed: Option<super::receipt_fragments::EncodedReceipt>,
    cause: Box<dyn std::error::Error + Send + Sync>,
}
impl std::fmt::Display for ReceiptDeliveryError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "receipt_delivery_failed: {}; {}; retained {} sealed bytes",
            self.original,
            self.cause,
            self.sealed
                .as_ref()
                .map_or(0, |receipt| receipt.bytes().len())
        )
    }
}
impl std::error::Error for ReceiptDeliveryError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(self.cause.as_ref())
    }
}

fn connection(stream: UnixStream, service: &Service, limits: Limits) -> io::Result<()> {
    let mut stream = Deadline {
        stream,
        end: deadline(limits.frame_timeout)?,
    };
    let mut cap = limits.max_frame;
    let mut greeted = false;
    // The Lys pass this connection presented, if any; every later request on
    // it is decided from it.
    let mut pass: Option<Presented> = None;
    loop {
        stream.end = deadline(limits.frame_timeout)?;
        let Some(bytes) = frame::read(&mut stream, cap)? else {
            return Ok(());
        };
        let response_limit = NonZeroUsize::new(cap.get() as usize)
            .ok_or_else(|| io::Error::other("invalid response frame limit"))?;
        let (id, hello, mutation, result) = match decode_request(&bytes)? {
            DecodedRequest::Request(request) if follows(&request, greeted) => {
                follow_head(
                    &mut stream,
                    service,
                    limits.frame_timeout,
                    cap,
                    &request,
                    &bytes,
                    pass.as_ref(),
                )?;
                continue;
            }
            DecodedRequest::Request(request) => {
                let hello = matches!(&request.operation, Operation::Hello(_));
                let mutation = MutationContext::from_operation(&request.operation);
                let agreement = match &request.operation {
                    Operation::Hello(agreement) => agreement.validate(limits.max_frame),
                    _ => Ok(cap),
                };
                let result = match agreement {
                    Err(failure) => Err(ResponseFailure::new(failure, mutation.clone())),
                    Ok(_) if !greeted && !hello => Err(ResponseFailure::new(
                        Failure::new("hello_required", "send hello before other requests"),
                        mutation.clone(),
                    )),
                    Ok(agreed) => {
                        cap = agreed;
                        let budget = NonZeroUsize::new(cap.get() as usize)
                            .ok_or_else(|| io::Error::other("invalid response frame limit"))?;
                        answer_agreed(
                            service,
                            request.operation,
                            &bytes,
                            &mut pass,
                            mutation.as_ref(),
                            budget,
                            limits.frame_timeout,
                        )
                    }
                };
                (request.id, hello, mutation, result)
            }
            DecodedRequest::Refused { id, failure } => {
                let hello = matches!(
                    failure.code.as_str(),
                    "typed_receipt_capability_missing"
                        | "receipt_protocol_mismatch"
                        | "receipt_frame_too_small"
                );
                let mutation = failure.mutation_context();
                (id, hello, mutation, Err(failure))
            }
        };
        let close = hello && result.is_err();
        if hello {
            greeted = result.is_ok();
        }
        if let Err(error) = &result
            && deliver_receipt(&mut stream, error, &id, cap)?
        {
            if close {
                return Ok(());
            }
            continue;
        }
        let response = response_text(&id, hello, cap, result, mutation, response_limit)?;
        stream.end = deadline(limits.frame_timeout)?;
        frame::write(&mut stream, response.get().as_bytes(), cap)?;
        if close {
            return Ok(());
        }
    }
}

/// The Lys pass a connection presented, and when it expires.
struct Presented {
    token: String,
    /// Seconds since the epoch, as the service read it from the pass; `None`
    /// only when its answer carried none.
    expires: Option<u64>,
}

/// Answers one request after the hello agreement: `session.pass` is presented
/// to the service and kept as this connection's pass when it is accepted;
/// every other operation is authorized against that pass before it runs.
fn answer_agreed(
    service: &Service,
    operation: Operation,
    bytes: &[u8],
    pass: &mut Option<Presented>,
    mutation: Option<&MutationContext>,
    budget: NonZeroUsize,
    timeout: Duration,
) -> Result<Box<serde_json::value::RawValue>, ResponseFailure> {
    if let Operation::Present(presented) = operation {
        let answered = service.present(&presented.pass);
        let expires = answered
            .as_ref()
            .ok()
            .and_then(|answer| answer.get("expires"))
            .and_then(serde_json::Value::as_u64);
        let answer = answered
            .and_then(|answer| serde_json::value::to_raw_value(&answer).map_err(Failure::storage))
            .map_err(|failure| ResponseFailure::new(failure, mutation.cloned()));
        if answer.is_ok() {
            *pass = Some(Presented {
                token: presented.pass,
                expires,
            });
        }
        answer
    } else {
        let token = pass.as_ref().map(|pass| pass.token.as_str());
        let words = std::str::from_utf8(bytes).ok();
        service
            .authorize(token, &operation, words)
            .map_err(|failure| ResponseFailure::new(failure, mutation.cloned()))
            .and_then(|()| match operation {
                // The runner acts with this connection's pass, as
                // session.pass is answered by this connection.
                Operation::DraftsRun(run) => service
                    .run_drafts_waiting(token, &run, timeout)
                    .and_then(|answer| {
                        serde_json::value::to_raw_value(&answer).map_err(Failure::storage)
                    })
                    .map_err(|failure| ResponseFailure::new(failure, mutation.cloned())),
                operation => service.dispatch_text_waiting(operation, budget, timeout),
            })
    }
}

/// Whether `request` is a follow this connection answers as a stream: one
/// sent after hello. Before hello it is refused `hello_required`, as any is.
const fn follows(request: &Request, greeted: bool) -> bool {
    greeted && matches!(&request.operation, Operation::Follow(_))
}

/// Answers `head.follow` as a stream of frames, every one with the request's
/// id (serve/follow.rs): once the connection's pass allows it, the head at
/// once and one frame per change of it, then the refusal that ends it; a
/// refused follow is answered by its refusal alone and sent no head.
///
/// The follow holds this connection's thread, and so its one client slot,
/// for as long as it lasts. It only writes, each frame under a fresh frame
/// deadline, so the idle read deadline never cuts it; the next request is
/// read under a fresh one. A write that fails means the client went away and
/// ends the connection. Socket readiness also wakes it as soon as the peer closes.
fn follow_head(
    stream: &mut Deadline,
    service: &Service,
    frame_timeout: Duration,
    cap: NonZeroU32,
    request: &Request,
    bytes: &[u8],
    pass: Option<&Presented>,
) -> io::Result<()> {
    let response_limit = NonZeroUsize::new(cap.get() as usize)
        .ok_or_else(|| io::Error::other("invalid response frame limit"))?;
    let id = request.id.as_str();
    let token = pass.map(|pass| pass.token.as_str());
    let words = std::str::from_utf8(bytes).ok();
    let ending = match (
        service.authorize(token, &request.operation, words),
        &request.operation,
    ) {
        (Err(failure), _) => failure,
        (Ok(()), Operation::Follow(asked)) => service.follow_socket(
            &stream.stream.try_clone()?,
            &asked.branch,
            pass.and_then(|pass| pass.expires),
            |head| {
                let result = text(head)?;
                let response = response_text(id, false, cap, Ok(result), None, response_limit)?;
                stream.end = deadline(frame_timeout)?;
                frame::write(&mut *stream, response.get().as_bytes(), cap)
            },
        )?,
        (Ok(()), _) => Failure::new("invalid_operation", "only head.follow is followed"),
    };
    let response = response_text(
        id,
        false,
        cap,
        Err(ResponseFailure::new(ending, None)),
        None,
        response_limit,
    )?;
    stream.end = deadline(frame_timeout)?;
    frame::write(stream, response.get().as_bytes(), cap)
}

/// Writes a refusal's sealed receipt as fragment frames. Answers whether the
/// refusal carried a receipt, in which case no plain response follows it.
fn deliver_receipt(
    stream: &mut Deadline,
    error: &ResponseFailure,
    id: &str,
    cap: NonZeroU32,
) -> io::Result<bool> {
    let sealed = error.failure().seal().map_err(|cause| {
        io::Error::other(ReceiptDeliveryError {
            original: error.clone(),
            sealed: None,
            cause: Box::new(cause),
        })
    })?;
    let Some(sealed) = sealed else {
        return Ok(false);
    };
    let frames = sealed.frames(id, cap.get()).map_err(|cause| {
        io::Error::other(ReceiptDeliveryError {
            original: error.clone(),
            sealed: None,
            cause: Box::new(cause),
        })
    })?;
    for body in frames {
        if let Err(cause) = frame::write(stream, &body, cap) {
            return Err(io::Error::other(ReceiptDeliveryError {
                original: error.clone(),
                sealed: Some(sealed),
                cause: Box::new(cause),
            }));
        }
    }
    Ok(true)
}

/// The one response frame for a request: the hello answer carries the agreed
/// frame cap, and an answer over the response limit becomes a refusal that fits.
fn response_text(
    id: &str,
    hello: bool,
    cap: NonZeroU32,
    result: Result<Box<serde_json::value::RawValue>, ResponseFailure>,
    mutation: Option<MutationContext>,
    response_limit: NonZeroUsize,
) -> io::Result<Box<serde_json::value::RawValue>> {
    let mutation = result
        .as_ref()
        .err()
        .and_then(ResponseFailure::mutation_context)
        .or(mutation);
    let response = match result {
        Ok(result) if hello => {
            let mut result: serde_json::Value =
                serde_json::from_str(result.get()).map_err(io::Error::other)?;
            result["max_frame"] = json!(cap);
            text(&json!({"id":id, "result":result}))?
        }
        Ok(result) => text(&Replied {
            id,
            result: &result,
        })?,
        Err(error) => text(&json!({"id":id, "error":error}))?,
    };
    let response = if response.get().len() <= response_limit.get() {
        response
    } else {
        text(
            &json!({"id":id,"error":ResponseFailure::after_dispatch(Failure::new("response_limit", "response exceeds frame limit; use a smaller page, or reconcile a mutation by its request ID"), mutation)}),
        )?
    };
    if response.get().len() > response_limit.get() {
        return Err(io::Error::other("frame limit cannot fit refusal envelope"));
    }
    Ok(response)
}

struct Deadline {
    stream: UnixStream,
    end: Instant,
}

fn deadline(timeout: Duration) -> io::Result<Instant> {
    Instant::now()
        .checked_add(timeout)
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "frame deadline overflow"))
}
impl Deadline {
    fn remaining(&self) -> io::Result<Duration> {
        self.end
            .checked_duration_since(Instant::now())
            .filter(|value| !value.is_zero())
            .ok_or_else(|| io::Error::new(io::ErrorKind::TimedOut, "frame deadline elapsed"))
    }

    /// macOS refuses a socket timeout with EINVAL once the peer has closed. A
    /// socket in that state never blocks, so the one call is made without the
    /// deadline: a read gets what is still buffered and then end of file, a
    /// write gets the peer's close. A call that would block means the peer is
    /// still there and the deadline could not be set, which is returned.
    fn without_deadline(
        &mut self,
        refused: io::Error,
        call: impl FnOnce(&mut UnixStream) -> io::Result<usize>,
    ) -> io::Result<usize> {
        if refused.kind() != io::ErrorKind::InvalidInput {
            return Err(refused);
        }
        self.stream.set_nonblocking(true)?;
        let done = call(&mut self.stream);
        self.stream.set_nonblocking(false)?;
        match done {
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("the frame deadline could not be set on a live connection: {refused}"),
            )),
            other => other,
        }
    }
}
impl Read for Deadline {
    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
        match self.stream.set_read_timeout(Some(self.remaining()?)) {
            Ok(()) => self.stream.read(bytes),
            Err(refused) => self.without_deadline(refused, |stream| stream.read(bytes)),
        }
    }
}
impl Write for Deadline {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        match self.stream.set_write_timeout(Some(self.remaining()?)) {
            Ok(()) => self.stream.write(bytes),
            Err(refused) => self.without_deadline(refused, |stream| stream.write(bytes)),
        }
    }
    fn flush(&mut self) -> io::Result<()> {
        self.stream.flush()
    }
}

/// A response whose result is text, written as it is (NB1).
#[derive(serde::Serialize)]
struct Replied<'a> {
    id: &'a str,
    result: &'a serde_json::value::RawValue,
}

/// `value` as one line of response text.
fn text<T: serde::Serialize>(value: &T) -> io::Result<Box<serde_json::value::RawValue>> {
    serde_json::value::to_raw_value(value).map_err(io::Error::other)
}

#[cfg(test)]
#[path = "transport_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "transport_follow_tests.rs"]
mod follow_tests;

#[cfg(test)]
#[path = "transport_admission_tests.rs"]
mod admission_tests;