haematite-cli 0.21.3

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

use std::fs;
use std::io::{self, Read, Write};
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

use serde_json::json;

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

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

/// Holds this listener's socket and optional published record until owned cleanup.
pub struct Listener {
    socket: UnixListener,
    path: PathBuf,
    identity: (u64, u64),
    entry: Option<super::registry::Publication>,
    // Retained until after Drop removes our socket; never unlink the lock anchor.
    _ownership: fs::File,
}

impl Listener {
    /// # Errors
    ///
    /// Propagates any error from `acquire`, `clear_stale`, `bind` or `set_permissions`.
    pub fn bind(path: &Path) -> io::Result<Self> {
        let ownership = super::socket_owner::acquire(path)?;
        super::socket_owner::clear_stale(path)?;
        let socket = UnixListener::bind(path)?;
        let setup = (|| {
            fs::set_permissions(path, fs::Permissions::from_mode(0o600))?;
            fs::symlink_metadata(path)
        })();
        let metadata = match setup {
            Ok(metadata) => metadata,
            Err(error) => {
                if let Err(cleanup) = fs::remove_file(path) {
                    eprintln!("haematite socket setup cleanup: {cleanup}");
                }
                return Err(error);
            }
        };
        Ok(Self {
            socket,
            path: path.to_owned(),
            identity: (metadata.dev(), metadata.ino()),
            entry: None,
            _ownership: ownership,
        })
    }

    /// Publishes the store only after this listener holds socket ownership.
    ///
    /// # Errors
    /// Returns a named publication refusal; callers must withhold readiness on failure.
    pub fn publish(
        &mut self,
        endpoint: &super::runtime::Endpoint,
    ) -> Result<(), super::registry::Refusal> {
        if endpoint.socket != self.path || self.entry.is_some() {
            return Err(super::registry::Refusal::new(
                "registry_publication_failed",
                "listener endpoint differs or already has an entry",
            ));
        }
        self.entry = Some(super::registry::Publication::publish(endpoint)?);
        Ok(())
    }

    /// Removes this listener's entry while it still holds socket ownership.
    ///
    /// # Errors
    /// Returns a named cleanup refusal without removing a replacement entry.
    pub fn finish(&mut self) -> Result<(), super::registry::Refusal> {
        if let Some(mut entry) = self.entry.take() {
            entry.finish()?;
        }
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn publish_in(
        &mut self,
        directory: &Path,
        entry: super::registry::Entry,
    ) -> Result<(), super::registry::Refusal> {
        if entry.socket != self.path || self.entry.is_some() {
            return Err(super::registry::Refusal::new(
                "registry_publication_failed",
                "listener endpoint differs or already has an entry",
            ));
        }
        self.entry = Some(super::registry::Publication::publish_in(directory, entry)?);
        Ok(())
    }

    /// # Errors
    ///
    /// Refuses invalid serving limits or reclaim activation; client failures are logged.
    pub fn run(&self, service: &Arc<Service>, limits: Limits) -> io::Result<()> {
        Self::accepting(self.socket.incoming(), service, limits, spawn_client)
    }

    fn accepting(
        incoming: impl Iterator<Item = io::Result<UnixStream>>,
        service: &Arc<Service>,
        limits: Limits,
        mut spawn: impl FnMut(UnixStream, Arc<Service>, Limits, Permit) -> io::Result<()>,
    ) -> io::Result<()> {
        let active = Self::begin_serving(service, limits)?;
        for incoming in incoming {
            let stream = match incoming {
                Ok(stream) => stream,
                Err(error) => {
                    let name = accept_error_name(&error);
                    eprintln!("haematite accept_failed: {name}: {error}");
                    if matches!(error.raw_os_error(), Some(23 | 24)) {
                        std::thread::sleep(Duration::from_millis(25));
                    }
                    continue;
                }
            };
            if let Err(count) = active.fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| {
                (count < limits.max_clients.get()).then(|| count + 1)
            }) {
                refuse_client(
                    &stream,
                    limits,
                    "busy_max_clients",
                    &format!("max_clients={}; active_clients={count}", limits.max_clients),
                );
                continue;
            }
            let permit = Permit(Arc::clone(&active));
            let service = Arc::clone(service);
            let refusal = match stream.try_clone() {
                Ok(refusal) => refusal,
                Err(error) => {
                    refuse_client(
                        &stream,
                        limits,
                        "client_spawn_failed",
                        &format!("socket_clone_failed: {error}"),
                    );
                    continue;
                }
            };
            if let Err(error) = spawn(stream, service, limits, permit) {
                refuse_client(&refusal, limits, "client_spawn_failed", &error.to_string());
            }
        }
        Ok(())
    }

    fn begin_serving(service: &Service, limits: Limits) -> io::Result<Arc<AtomicUsize>> {
        if limits.frame_timeout.is_zero() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "frame timeout must be positive",
            ));
        }
        service
            .activate_reclaim()
            .map_err(|error| io::Error::other(format!("reclaim_activation_failed: {error}")))?;
        Ok(Arc::new(AtomicUsize::new(0)))
    }
}

impl Drop for Listener {
    fn drop(&mut self) {
        if let Err(error) = self.finish() {
            eprintln!("{error}");
        }
        if let Ok(metadata) = fs::symlink_metadata(&self.path)
            && (metadata.dev(), metadata.ino()) == self.identity
            && let Err(error) = fs::remove_file(&self.path)
        {
            eprintln!("haematite socket cleanup: {error}");
        }
    }
}

/// 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}")
}

fn accept_error_name(error: &io::Error) -> String {
    match error.raw_os_error() {
        Some(24) => "EMFILE".to_owned(),
        Some(23) => "ENFILE".to_owned(),
        _ if error.kind() == io::ErrorKind::ConnectionAborted => "ECONNABORTED".to_owned(),
        _ => format!("{:?}", error.kind()),
    }
}

fn refusal_id(stream: &UnixStream, limits: Limits) -> io::Result<String> {
    stream.set_read_timeout(Some(limits.frame_timeout.min(Duration::from_millis(250))))?;
    let cap = limits.max_frame.min(NonZeroU32::MIN.saturating_add(65_535));
    let bytes = frame::read(&mut &*stream, cap)?
        .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer sent no request id"))?;
    let request: serde_json::Value = serde_json::from_slice(&bytes)
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
    request
        .get("id")
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "request has no string id"))
}

fn refuse_client(stream: &UnixStream, limits: Limits, code: &str, message: &str) {
    let (id, message) = match refusal_id(stream, limits) {
        Ok(id) => (id, message.to_owned()),
        Err(error) => (
            String::new(),
            format!(
                "{message}; request_id_unavailable: {:?}: {error}",
                error.kind()
            ),
        ),
    };
    let result = (|| {
        let response = text(
            &json!({"id": id, "error": ResponseFailure::new(Failure::new(code, message.clone()), None)}),
        )?;
        stream.set_write_timeout(Some(limits.frame_timeout.min(Duration::from_millis(250))))?;
        frame::write(&mut &*stream, response.get().as_bytes(), limits.max_frame)
    })();
    match result {
        Ok(()) => eprintln!("haematite {code}: {message}"),
        Err(error) => eprintln!("haematite {code}: {message}; refusal_delivery_failed: {error}"),
    }
}

fn spawn_client(
    stream: UnixStream,
    service: Arc<Service>,
    limits: Limits,
    permit: Permit,
) -> io::Result<()> {
    std::thread::Builder::new()
        .name("haematite-client".to_owned())
        .spawn(move || {
            let _permit = permit;
            if let Err(error) = connection(stream, &service, limits) {
                eprintln!("{}", ending(&error));
            }
        })
        .map(|_| ())
}

struct Permit(Arc<AtomicUsize>);
impl Drop for Permit {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::AcqRel);
    }
}

#[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;