haematite-cli 0.21.8

#![cfg(unix)]
#![forbid(unsafe_code)]

use std::error::Error;
use std::num::NonZeroU32;
use std::os::unix::{fs::PermissionsExt, net::UnixStream};
use std::path::Path;
use std::time::{Duration, Instant};

use haematite_cli::serve::frame;
use haematite_cli::serve::protocol::{PROTOCOL, RECEIPT_CAPABILITY};
use serde_json::{Value, json};

type TestResult = Result<(), Box<dyn Error>>;

#[path = "support/served_child.rs"]
mod served_child;
use served_child::Running;

fn start(config: &Path, create: bool) -> Result<Running, Box<dyn Error>> {
    Running::start(config, create)
}

fn config(dir: &Path) -> Result<std::path::PathBuf, Box<dyn Error>> {
    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))?;
    let path = dir.join("serve.json");
    std::fs::write(
        &path,
        serde_json::to_vec(&json!({
            "data_dir":dir.join("database"), "socket":dir.join("serve.sock"),
            "name":dir.file_name().and_then(|name| name.to_str()).ok_or("fixture name is not text")?,
            "branch_cache_bytes_per_shard":1_048_576, "max_frame":1_048_576,
            "max_clients":4, "max_page":20, "timeout_ms":2000,
            "create":{"shards":2,"cache_bytes_total":2_097_152,"executor_threads":2}
        }))?,
    )?;
    Ok(path)
}

fn exchange(stream: &mut UnixStream, operation: Value) -> Result<Value, Box<dyn Error>> {
    let bound = NonZeroU32::new(1_048_576).ok_or("invalid bound")?;
    let mut request = json!({"id":"test"});
    request["operation"] = operation;
    frame::write(stream, &serde_json::to_vec(&request)?, bound)?;
    let bytes = frame::read(stream, bound)?.ok_or("server closed before response")?;
    let response: Value = serde_json::from_slice(&bytes)?;
    assert_eq!(response["id"], "test");
    Ok(response)
}

/// The hello a peer sends under BUSYSHARDNODEADLINE-001 R12: it agrees the
/// protocol and the typed receipt capability, and offers the frame it reads.
fn hello() -> Value {
    json!({"method":"hello", "params":{
        "protocol":PROTOCOL, "capabilities":[RECEIPT_CAPABILITY], "max_frame":1_048_576,
    }})
}

fn connect(dir: &Path) -> Result<UnixStream, Box<dyn Error>> {
    let mut stream = UnixStream::connect(dir.join("serve.sock"))?;
    stream.set_read_timeout(Some(Duration::from_secs(3)))?;
    stream.set_write_timeout(Some(Duration::from_secs(3)))?;
    assert_eq!(
        exchange(&mut stream, hello())?["result"]["protocol"],
        "haem/1"
    );
    Ok(stream)
}

fn head(stream: &mut UnixStream) -> Result<Value, Box<dyn Error>> {
    let response = exchange(
        stream,
        json!({"method":"branches.list","params":{"limit":20}}),
    )?;
    let rows = response["result"]["entries"]
        .as_array()
        .ok_or("missing branch list")?;
    Ok(rows
        .iter()
        .find(|row| row["branch"] == "main")
        .ok_or("main missing")?["version"]
        .clone())
}

fn mutation(id: &str, expected: Value, change: Value) -> Value {
    let mut operation =
        json!({"method":"mutation.apply","params":{"request_id":id,"branch":"main"}});
    operation["params"]["expected"] = expected;
    operation["params"]["change"] = change;
    operation
}

fn wait_published(client: &mut UnixStream, previous: &Value) -> TestResult {
    let deadline = Instant::now() + Duration::from_secs(5);
    while head(client)? == *previous {
        if Instant::now() >= deadline {
            return Err("original mutation did not publish before deadline".into());
        }
        std::thread::yield_now();
    }
    Ok(())
}

fn wait_receipt(client: &mut UnixStream, request_id: &str) -> Result<Value, Box<dyn Error>> {
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let reply = exchange(
            client,
            json!({"method":"mutation.receipt","params":{"branch":"main","request_id":request_id}}),
        )?;
        if reply["result"].is_object() {
            return Ok(reply);
        }
        if !reply["error"].is_null() && reply["error"]["code"] != "busy" {
            return Err(format!("receipt refused: {reply}").into());
        }
        if Instant::now() >= deadline {
            return Err("receipt did not appear within deadline".into());
        }
        std::thread::yield_now();
    }
}

#[test]
fn external_client_writes_reads_and_reconciles_after_process_restart() -> TestResult {
    let dir = tempfile::tempdir()?;
    let config = config(dir.path())?;
    let process = start(&config, true)?;
    let mut client = connect(dir.path())?;
    let expected = head(&mut client)?;
    let declare = mutation(
        "declare",
        expected,
        json!({"kind":"collection.put","value":{"collection":"records","schema":{"type":"object","required":["name"]}}}),
    );
    assert!(exchange(&mut client, declare)?["result"].is_object());
    let expected = head(&mut client)?;
    let first = mutation(
        "first",
        expected,
        json!({"kind":"document.put","value":{"collection":"records","document_id":"one","document":{"name":"earlier","extra":true}}}),
    );
    // Observe commit through another connection, then discard the original
    // connection without reading ANY mutation reply bytes. Sending alone does
    // not prove receipt, so closing earlier cannot assert a committed write.
    let mut abandoned = connect(dir.path())?;
    frame::write(
        &mut abandoned,
        &serde_json::to_vec(&json!({"id":"dropped","operation":first}))?,
        NonZeroU32::new(1_048_576).ok_or("invalid bound")?,
    )?;
    wait_published(&mut client, &first["params"]["expected"])?;
    drop(abandoned);
    let first_response = wait_receipt(&mut client, "first")?;
    let expected = head(&mut client)?;
    let later = mutation(
        "later",
        expected,
        json!({"kind":"document.put","value":{"collection":"records","document_id":"one","document":{"name":"later"}}}),
    );
    assert!(exchange(&mut client, later)?["result"].is_object());
    drop(client);
    drop(process); // Abrupt process death also leaves a stale socket entry.
    let _restarted = start(&config, false)?;
    let mut client = connect(dir.path())?;
    assert_eq!(
        exchange(&mut client, first)?["result"],
        first_response["result"]
    );
    let read = exchange(
        &mut client,
        json!({"method":"documents.get","params":{"selection":{"branch":"main"},"collection":"records","document_id":"one"}}),
    )?;
    assert_eq!(
        read["result"]["document"]["raw_json"],
        r#"{"name":"later"}"#
    );
    Ok(())
}

#[test]
fn endpoint_refuses_schema_and_unpinned_continuations() -> TestResult {
    let dir = tempfile::tempdir()?;
    let config = config(dir.path())?;
    let _process = start(&config, true)?;
    let mut client = connect(dir.path())?;
    let expected = head(&mut client)?;
    let declare = mutation(
        "schema",
        expected,
        json!({"kind":"collection.put","value":{"collection":"records","schema":{"required":["name"]}}}),
    );
    assert!(exchange(&mut client, declare)?["result"].is_object());
    let expected = head(&mut client)?;
    let invalid = mutation(
        "invalid",
        expected.clone(),
        json!({"kind":"document.put","value":{"collection":"records","document_id":"one","document":{"extra":true}}}),
    );
    assert_eq!(exchange(&mut client, invalid)?["error"]["path"], "/name");
    assert_eq!(head(&mut client)?, expected);
    let page = exchange(
        &mut client,
        json!({"method":"documents.list","params":{"selection":{"branch":"main"},"collection":"records","after":"one","limit":1}}),
    )?;
    assert_eq!(page["error"]["code"], "version_required");
    Ok(())
}

#[test]
fn two_socket_readers_keep_a_version_while_a_writer_advances() -> TestResult {
    let dir = tempfile::tempdir()?;
    let config = config(dir.path())?;
    let _process = start(&config, true)?;
    let mut writer = connect(dir.path())?;
    let expected = head(&mut writer)?;
    let declare = mutation(
        "schema",
        expected,
        json!({"kind":"collection.put","value":{"collection":"records","schema":null}}),
    );
    assert!(exchange(&mut writer, declare)?["result"].is_object());
    let expected = head(&mut writer)?;
    let initial = mutation(
        "initial",
        expected,
        json!({"kind":"document.put","value":{"collection":"records","document_id":"one","document":{"number":1}}}),
    );
    assert!(exchange(&mut writer, initial)?["result"].is_object());
    let pinned = head(&mut writer)?;
    let readers = [connect(dir.path())?, connect(dir.path())?];
    std::thread::scope(|scope| -> TestResult {
        let (finished, completions) = std::sync::mpsc::channel();
        let mut commands = Vec::new();
        let mut workers = Vec::new();
        for mut reader in readers {
            let (start, requests) = std::sync::mpsc::channel();
            commands.push(start);
            let pinned = pinned.clone();
            let finished = finished.clone();
            workers.push(scope.spawn(move || -> Result<(), String> {
                while requests.recv_timeout(Duration::from_secs(5)).is_ok() {
                    let response = exchange(&mut reader, json!({"method":"documents.list","params":{
                        "selection":{"branch":"main","version":pinned},"collection":"records","limit":1
                    }})).map_err(|error| error.to_string());
                    let result = response.and_then(|response| {
                        if response["result"]["entries"][0]["raw_json"] == r#"{"number":1}"# { Ok(()) }
                        else { Err(format!("unexpected pinned response: {response}")) }
                    });
                    finished.send(result).map_err(|error| error.to_string())?;
                }
                Ok(())
            }));
        }
        for number in 2..5 {
            for start in &commands {
                start.send(())?;
            }
            let expected = head(&mut writer)?;
            let write = mutation(
                &format!("update-{number}"),
                expected,
                json!({"kind":"document.put","value":{
                    "collection":"records","document_id":"one","document":{"number":number}
                }}),
            );
            assert!(exchange(&mut writer, write)?["result"].is_object());
            for _ in 0..2 {
                completions
                    .recv_timeout(Duration::from_secs(5))?
                    .map_err(std::io::Error::other)?;
            }
        }
        drop(commands);
        for worker in workers {
            worker
                .join()
                .map_err(|_| "reader panicked")?
                .map_err(std::io::Error::other)?;
        }
        Ok(())
    })?;
    Ok(())
}

#[test]
fn client_admission_refuses_excess_without_disturbing_existing_readers() -> TestResult {
    let dir = tempfile::tempdir()?;
    let config = config(dir.path())?;
    let _process = start(&config, true)?;
    let mut admitted = (0..4)
        .map(|_| connect(dir.path()))
        .collect::<Result<Vec<_>, _>>()?;
    let mut excess = UnixStream::connect(dir.path().join("serve.sock"))?;
    excess.set_read_timeout(Some(Duration::from_secs(3)))?;
    excess.set_write_timeout(Some(Duration::from_secs(3)))?;
    let refusal = exchange(&mut excess, hello())?;
    assert_eq!(refusal["error"]["code"], "busy_max_clients");
    let message = refusal["error"]["message"]
        .as_str()
        .ok_or("refusal has no message")?;
    assert!(
        message.contains("max_clients=4") && message.contains("active_clients=4"),
        "{message}"
    );
    assert!(
        frame::read(
            &mut excess,
            NonZeroU32::new(1_048_576).ok_or("invalid bound")?
        )?
        .is_none()
    );
    assert_eq!(
        exchange(&mut admitted[0], hello())?["result"]["protocol"],
        "haem/1"
    );
    Ok(())
}

#[test]
fn oversized_frame_disconnects_only_its_client_before_body_read() -> TestResult {
    use std::io::{Read, Write};
    let dir = tempfile::tempdir()?;
    let config = config(dir.path())?;
    let _process = start(&config, true)?;
    let mut malformed = UnixStream::connect(dir.path().join("serve.sock"))?;
    malformed.set_read_timeout(Some(Duration::from_secs(2)))?;
    malformed.write_all(&1_048_577_u32.to_be_bytes())?;
    let result = malformed.read(&mut [0_u8; 1]);
    assert!(
        matches!(result, Ok(0))
            || result
                .as_ref()
                .is_err_and(|error| error.kind() == std::io::ErrorKind::ConnectionReset)
    );
    let mut healthy = connect(dir.path())?;
    assert!(head(&mut healthy)?.is_object());
    Ok(())
}