haematite_cli/serve/

cli.rs

//! Explicit operator startup. Reading configuration never installs a service.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use haematite::store::NodeCacheBudget;

use super::{
    embed,
    query::{Setting, probe},
    service::Service,
    storage::{Storage, UndatedGeneration},
    transport::{Limits, Listener},
};

/// Checks ownership of a resolved default name endpoint without changing it.
///
/// The expected uid is supplied by the caller's validated OS runtime context.
/// A successful probe lock is released before returning stopped; pids and
/// socket connection results do not participate in this observation.
///
/// # Errors
/// Returns `name_owner_probe_failed` with the anchor's cause when its location,
/// owner, mode, identity or OS lock cannot be checked or released.
pub fn name_is_live(default_endpoint: &std::path::Path, expected_uid: u32) -> Result<bool, String> {
    super::socket_owner::name_is_live(default_endpoint, expected_uid).map_err(|error| {
        format!(
            "name_owner_probe_failed: {}: {error}",
            default_endpoint.display()
        )
    })
}

const USAGE: &str = "usage: haematite serve --config <file.json> [--create] [--long-wait-interval <duration>] [--exit-on-stdin-close]\n\
--create requires a new database directory; omit it when restarting.\n\
An explicit socket's parent directory must already exist with mode 0700.\n\
An omitted socket uses the private OS runtime namespace. No auto-launch is installed.";

const DEFAULT_LONG_WAIT_INTERVAL: Duration = Duration::from_secs(30);

struct ServeOptions {
    path: PathBuf,
    create: bool,
    exit_on_stdin_close: bool,
    long_wait_interval: Duration,
}

fn long_wait_help() -> String {
    format!(
        "--long-wait-interval <duration> is the interval between long-wait warnings, not a deadline. \
         default: {}s; use an integer followed by ns, us, ms, s, m or h.",
        DEFAULT_LONG_WAIT_INTERVAL.as_secs()
    )
}

#[path = "cli_config.rs"]
mod config;
#[cfg(test)]
use config::lys_gate;
pub use config::{Config, Creation, LysConfig};
use config::{creation_config, lys_parts, served_through_lys};

fn claim_name(
    config: &Config,
    endpoint: &super::runtime::Endpoint,
    create: bool,
) -> Result<std::fs::File, String> {
    let name_endpoint = if config.socket.is_some() {
        #[cfg(test)]
        let resolved = match endpoint_tests::input() {
            Some((root, uid)) => super::runtime::resolve_input(
                &config.data_dir,
                Some(&endpoint.name),
                None,
                create,
                &root,
                uid,
            ),
            None => super::runtime::resolve(&config.data_dir, Some(&endpoint.name), None, create),
        };
        #[cfg(not(test))]
        let resolved =
            super::runtime::resolve(&config.data_dir, Some(&endpoint.name), None, create);
        Some(resolved.map_err(|error| format!("bind service socket: {error}"))?)
    } else {
        None
    };
    let name_socket = name_endpoint
        .as_ref()
        .map_or(&endpoint.socket, |named| &named.socket);
    let namespace = name_socket
        .parent()
        .ok_or("bind service socket: name endpoint needs a parent")?;
    let name_uid = {
        use std::os::unix::fs::MetadataExt;
        std::fs::symlink_metadata(namespace)
            .map_err(|error| format!("bind service socket: {error}"))?
            .uid()
    };
    super::socket_owner::acquire_name(name_socket, name_uid).map_err(|error| {
        // The name lock is held: another running serve registered a store of
        // this name. Every other cause keeps the OS's own words.
        if error.kind() == std::io::ErrorKind::WouldBlock {
            format!(
                "bind service socket: the store name \"{}\" is already served by another running haematite serve ({}); give this data folder another name",
                endpoint.name,
                name_socket.display()
            )
        } else {
            format!("bind service socket: {error}")
        }
    })
}

/// The data directory and socket are absolute, the timeout fits the
/// monotonic clock, and the socket's parent is one this user may bind in;
/// the timeout is answered.
fn check_paths_and_timeout(config: &Config) -> Result<Duration, String> {
    if !config.data_dir.is_absolute()
        || config
            .socket
            .as_ref()
            .is_some_and(|socket| !socket.is_absolute())
    {
        return Err("data_dir and socket must be absolute paths".to_owned());
    }
    let timeout = Duration::from_millis(config.timeout_ms.get());
    if std::time::Instant::now().checked_add(timeout).is_none() {
        return Err("timeout cannot be represented by the monotonic clock".to_owned());
    }
    if let Some(socket) = config.socket.as_ref() {
        super::socket_owner::check_parent(socket)
            .map_err(|error| format!("bind service socket: {error}"))?;
    }
    Ok(timeout)
}

/// # Errors
///
/// Propagates any error from `parse`.
pub fn run(args: impl Iterator<Item = String>) -> Result<(), String> {
    let Some(options) = parse(args)? else {
        println!("{USAGE}\n{}", long_wait_help());
        return Ok(());
    };
    let config = read_config(&options.path)?;
    // Resolve the complete group before opening a store or binding a socket.
    let retention_limits = config.append_prune_limits()?;
    let query_settings = config.query_settings()?;
    let changes_settings = config.changes_settings()?;
    let release = config.retention.settings()?;
    // Read now, so a wrong key stops the service.
    let embedding = config.embed_settings()?;
    let (gate, placer) = lys_parts(config.lys.as_ref())?;
    let restart = embed::settings::Restart::of(
        config.embed_restart_ms,
        config.embed_restart_most_ms,
        embedding
            .as_ref()
            .map_or(Duration::ZERO, |settings| settings.ready),
    )?;
    let timeout = check_paths_and_timeout(&config)?;
    let endpoint = config
        .endpoint(options.create)
        .map_err(|error| error.to_string())?;
    let name_ownership = claim_name(&config, &endpoint, options.create)?;
    // A worker that cannot start does not stop the service: every other verb
    // is served, and the refusal is kept for `query.run` to answer by name.
    let query = query_settings.map(|settings| probe::prepare(settings, config.max_frame));
    if let Some(Err(refusal)) = query.as_ref().map(probe::Prepared::ready) {
        eprintln!("haem/1 query unavailable: {refusal}");
    }
    // Only the leaf-log mark is refused here; a missing or unreadable config
    // is left to the storage checks below, which answer as they always have.
    if let Err(error @ haematite::DatabaseError::LeafLogStore { .. }) =
        haematite::Database::check_ordinary_store(&config.data_dir)
    {
        return Err(error.to_string());
    }
    let branch_cache = NodeCacheBudget::Bytes(config.branch_cache_bytes_per_shard);
    let kept = config.kept_limit()?;
    let storage = if options.create {
        let creation = config
            .create
            .ok_or("--create needs a create section in the configuration")?;
        Storage::create_kept(
            creation_config(config.data_dir, &creation),
            branch_cache,
            kept,
            options.long_wait_interval,
        )
    } else {
        Storage::open_kept(
            &config.data_dir,
            branch_cache,
            kept,
            options.long_wait_interval,
        )
    }
    .map_err(|error| error.to_string())?;
    if options.create {
        endpoint
            .check_created()
            .map_err(|error| error.to_string())?;
    }
    let mut service =
        Service::with_query(storage, config.max_page, query).map_err(|error| error.to_string())?;
    service.changes = changes_settings;
    service.append_prune_limits = retention_limits;
    service = served_through_lys(service, gate, placer)?;
    service
        .start_release(release)
        .map_err(|error| error.to_string())?;
    // The worker starts on the embedding thread, off the serving path: until
    // it answers, text search is refused as `embed_worker_starting`.
    if let Some(settings) = embedding {
        let bounds = super::embedding::bounds(&settings);
        let writer_wait = settings.writer_wait;
        let apart = !matches!(config.embed_question_worker, Setting::Present(false));
        let questions = apart.then(|| super::embedding::embedder(settings.clone()));
        service
            .start_embedding_apart(
                super::embedding::embedder(settings),
                questions,
                bounds,
                writer_wait,
                restart,
            )
            .map_err(|error| error.to_string())?;
    }
    let service = Arc::new(service);
    service
        .activate_reclaim()
        .map_err(|error| format!("reclaim activation: {error}"))?;
    let served = serve_listener(
        &endpoint,
        &service,
        Limits {
            max_frame: config.max_frame,
            max_clients: config.max_clients,
            frame_timeout: timeout,
        },
        options.exit_on_stdin_close,
    );
    drop(name_ownership);
    served
}

fn serve_listener(
    endpoint: &super::runtime::Endpoint,
    service: &Arc<Service>,
    limits: Limits,
    exit_on_stdin_close: bool,
) -> Result<(), String> {
    let mut listener = Listener::bind(&endpoint.socket)
        .map_err(|error| format!("bind service socket: {error}"))?;
    listener
        .publish(endpoint)
        .map_err(|error| error.to_string())?;
    let watch = exit_on_stdin_close
        .then(super::stdin_close::Watch::start)
        .transpose()
        .map_err(|error| format!("stdin_watch_failed: {error}"))?;
    eprintln!("haem/1 ready on {}", endpoint.socket.display());
    let served = watch
        .map_or_else(
            || listener.run(service, limits),
            |watch| listener.run_until(service, limits, watch),
        )
        .map_err(|error| format!("serve: {error}"));
    let cleanup = listener.finish().map_err(|error| error.to_string());
    // The name remains owned while registry and socket cleanup still run.
    drop(listener);
    match (served, cleanup) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
        (Err(served), Err(cleanup)) => Err(format!("{served}; {cleanup}")),
    }
}

const RELEASE_USAGE: &str = "usage: haematite release date --config <file.json> --branch <name> \
--generation <created> --at <nanoseconds> [--long-wait-interval <duration>]\n\
Dates a generation discarded before its marker was written, as discarded at --at, so release \
takes its rows by keep.days from then. --generation is the record's creation time its rows are \
keyed by. Offline: the service must be stopped, as the store has one writer.";

struct DateOptions {
    branch: String,
    generation: u64,
    at: u64,
    long_wait_interval: Duration,
}

#[path = "cli/branch_rename.rs"]
mod branch_rename;
pub use branch_rename::{BRANCH_RENAME_USAGE, branch_rename};

/// DR-12: `haematite release date`, which dates a generation discarded before DR-1.
///
/// The operator's way out ([`Storage::date_undated_generation`], DR-11).
/// Offline: it opens the store as its one writer, so a running service holds
/// it off. Prints the marker's last version, `generation.sequence`.
///
/// # Errors
/// A usage error, the configuration's, the store's open refusal, or the
/// request's refusal by name (`branch_live`, `generation_marked`,
/// `generation_empty`).
pub fn release(mut args: impl Iterator<Item = String>) -> Result<(), String> {
    if args.next().as_deref() != Some("date") {
        return Err(RELEASE_USAGE.to_owned());
    }
    let mut args = args.peekable();
    if args.peek().is_some_and(|arg| arg == "--help") {
        if args.count() != 1 {
            return Err(RELEASE_USAGE.to_owned());
        }
        println!("{RELEASE_USAGE}\n{}", long_wait_help());
        return Ok(());
    }
    let (path, options) = parse_date(args)?;
    let config = read_config(&path)?;
    let branch_cache = NodeCacheBudget::Bytes(config.branch_cache_bytes_per_shard);
    let mut storage = Storage::open_kept(
        &config.data_dir,
        branch_cache,
        config.kept_limit()?,
        options.long_wait_interval,
    )
    .map_err(|error| error.to_string())?;
    let request = UndatedGeneration {
        branch: options.branch,
        generation: options.generation,
        at: options.at,
    };
    let last = storage
        .date_undated_generation(&request)
        .map_err(|failure| failure.to_string())?;
    println!("{}.{}", last.generation, last.sequence);
    Ok(())
}

/// The release request and its optional warning interval, each argument once;
/// the two record times are decimal nanoseconds.
fn parse_date(mut args: impl Iterator<Item = String>) -> Result<(PathBuf, DateOptions), String> {
    let (mut path, mut branch, mut generation, mut at) = (None, None, None, None);
    let mut long_wait_interval = None;
    while let Some(arg) = args.next() {
        let value = args.next().ok_or(RELEASE_USAGE)?;
        match arg.as_str() {
            "--config" if path.is_none() => path = Some(PathBuf::from(value)),
            "--branch" if branch.is_none() => branch = Some(value),
            "--generation" if generation.is_none() => generation = Some(decimal(&arg, &value)?),
            "--at" if at.is_none() => at = Some(decimal(&arg, &value)?),
            "--long-wait-interval" if long_wait_interval.is_none() => {
                long_wait_interval = Some(parse_long_wait_interval(&value)?);
            }
            _ => {
                return Err(format!(
                    "unknown or repeated argument {arg:?}\n{RELEASE_USAGE}"
                ));
            }
        }
    }
    let request = DateOptions {
        long_wait_interval: long_wait_interval.unwrap_or(DEFAULT_LONG_WAIT_INTERVAL),
        branch: branch.ok_or(RELEASE_USAGE)?,
        generation: generation.ok_or(RELEASE_USAGE)?,
        at: at.ok_or(RELEASE_USAGE)?,
    };
    Ok((path.ok_or(RELEASE_USAGE)?, request))
}

fn decimal(arg: &str, value: &str) -> Result<u64, String> {
    value
        .parse()
        .map_err(|error| format!("{arg} {value:?}: {error}\n{RELEASE_USAGE}"))
}

fn read_config(path: &std::path::Path) -> Result<Config, String> {
    let bytes = std::fs::read(path).map_err(|error| format!("read service config: {error}"))?;
    serde_json::from_slice(&bytes).map_err(|error| format!("service config: {error}"))
}

fn parse_long_wait_interval(value: &str) -> Result<Duration, String> {
    let invalid = || {
        format!(
            "invalid --long-wait-interval {value:?}: {}",
            long_wait_help()
        )
    };
    let end = value
        .find(|character: char| !character.is_ascii_digit())
        .ok_or_else(invalid)?;
    let (number, unit) = value.split_at(end);
    let amount: u64 = number.parse().map_err(|_| invalid())?;
    let interval = match unit {
        "ns" => Duration::from_nanos(amount),
        "us" => Duration::from_micros(amount),
        "ms" => Duration::from_millis(amount),
        "s" => Duration::from_secs(amount),
        "m" => Duration::from_secs(amount.checked_mul(60).ok_or_else(invalid)?),
        "h" => Duration::from_secs(amount.checked_mul(3_600).ok_or_else(invalid)?),
        _ => return Err(invalid()),
    };
    super::storage::runtime_options(interval).map_err(|error| error.to_string())?;
    Ok(interval)
}

fn parse(args: impl Iterator<Item = String>) -> Result<Option<ServeOptions>, String> {
    let mut args = args.peekable();
    if args.peek().is_some_and(|arg| arg == "--help") {
        return if args.count() == 1 {
            Ok(None)
        } else {
            Err(USAGE.to_owned())
        };
    }
    let mut path = None;
    let mut create = false;
    let mut exit_on_stdin_close = false;
    let mut long_wait_interval = None;
    while let Some(arg) = args.next() {
        match arg.as_str() {
            "--config" if path.is_none() => path = Some(PathBuf::from(args.next().ok_or(USAGE)?)),
            "--create" if !create => create = true,
            "--exit-on-stdin-close" if !exit_on_stdin_close => exit_on_stdin_close = true,
            "--long-wait-interval" if long_wait_interval.is_none() => {
                long_wait_interval = Some(parse_long_wait_interval(&args.next().ok_or(USAGE)?)?);
            }
            _ => return Err(format!("unknown or repeated argument {arg:?}\n{USAGE}")),
        }
    }
    Ok(Some(ServeOptions {
        path: path.ok_or(USAGE)?,
        create,
        exit_on_stdin_close,
        long_wait_interval: long_wait_interval.unwrap_or(DEFAULT_LONG_WAIT_INTERVAL),
    }))
}

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

#[cfg(test)]
#[path = "cli_lys_tests.rs"]
mod lys_tests;

#[cfg(test)]
#[path = "cli_long_wait_tests.rs"]
mod long_wait_tests;

#[cfg(test)]
#[path = "cli_endpoint_tests.rs"]
pub(crate) mod endpoint_tests;