haematite-cli 0.21.7

use std::io;
use std::num::NonZeroU32;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};

use crate::serve::protocol::{BranchPage, Operation, Version};
use crate::serve::service::Service;
use crate::serve::storage::{Failure, Storage};

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

const LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
/// The presented pass's expiry, in seconds since the epoch.
const EXPIRES: u64 = 2_000_000_000;
/// The bound on waiting for a frame that is due; a frame never comes late
/// unless the follow is broken, so this only stops a broken test hanging.
const WAIT: Duration = Duration::from_secs(30);

fn with_service(test: impl FnOnce(&Service) -> TestResult) -> TestResult {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let service = Service::new(storage, LIMIT)?;
    test(&service)
}

fn head(service: &Service, branch: &str) -> Result<Version, Box<dyn std::error::Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: LIMIT,
    }))?;
    let row = answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .find(|row| row["branch"] == branch)
        .ok_or_else(|| format!("branch {branch} missing"))?;
    Ok(serde_json::from_value(row["version"].clone())?)
}

/// Apply `change` to `branch` at its head, as request `request_id`.
fn mutate(service: &Service, branch: &str, request_id: &str, change: &Value) -> TestResult {
    let expected = head(service, branch)?;
    let operation: Operation = serde_json::from_value(json!({
        "method": "mutation.apply",
        "params": {"request_id": request_id, "branch": branch, "expected": expected, "change": change},
    }))?;
    service.dispatch(operation)?;
    Ok(())
}

fn collection(name: &str) -> Value {
    json!({"kind": "collection.put", "value": {"collection": name, "schema": null}})
}

fn fork_trial(service: &Service) -> TestResult {
    let from = head(service, "main")?;
    mutate(
        service,
        "main",
        "fork-trial",
        &json!({"kind": "branch.fork", "value": {"name": "trial", "from": from}}),
    )
}

fn at(seconds: u64) -> SystemTime {
    UNIX_EPOCH + Duration::from_secs(seconds)
}

/// The head frame a follower of `branch` sends at `version`.
fn frame(branch: &str, version: Version) -> Value {
    json!({"branch": branch, "version": version})
}

/// Follow `branch` on its own thread with a pass expiring at [`EXPIRES`],
/// read by a clock that stands before it until the pass is made late. `steps`
/// gets the frames. Then the pass is made late and main is published, which
/// ends a follower still waiting, so a failed step can never leave the test
/// hanging. Answers the steps' outcome and the refusal that ended the follow.
fn with_follower(
    service: &Service,
    branch: &str,
    steps: impl FnOnce(&mpsc::Receiver<Value>) -> TestResult,
) -> Result<(TestResult, Failure, Vec<Value>), Box<dyn std::error::Error>> {
    let late = AtomicBool::new(false);
    let (sent, frames) = mpsc::channel();
    std::thread::scope(|scope| {
        let follower = scope.spawn(|| {
            service.follow(
                branch,
                Some(EXPIRES),
                || {
                    if late.load(Ordering::SeqCst) {
                        at(EXPIRES)
                    } else {
                        at(EXPIRES - 600)
                    }
                },
                move |head: &Value| {
                    sent.send(head.clone())
                        .map_err(|error| io::Error::new(io::ErrorKind::BrokenPipe, error))
                },
            )
        });
        let outcome = steps(&frames);
        late.store(true, Ordering::SeqCst);
        mutate(service, "main", "wake", &collection("woken"))?;
        let ended = follower.join().map_err(|_| "the follower panicked")??;
        let after: Vec<Value> = frames.try_iter().collect();
        Ok((outcome, ended, after))
    })
}

/// Fails while a follower misses its own branch's publish, is sent a frame
/// for another branch's, or sends one before the head it starts from: it gets
/// the head at once, then exactly one frame per publish of its branch.
#[test]
fn a_follower_gets_its_head_then_one_frame_per_publish_of_its_branch() -> TestResult {
    with_service(|service| {
        fork_trial(service)?;
        let (outcome, ended, after) = with_follower(service, "main", |frames| {
            assert_eq!(
                frames.recv_timeout(WAIT)?,
                frame("main", head(service, "main")?)
            );
            mutate(service, "main", "one", &collection("one"))?;
            assert_eq!(
                frames.recv_timeout(WAIT)?,
                frame("main", head(service, "main")?)
            );
            // Another branch's publish wakes the follower and sends nothing:
            // the next frame is main's next publish, not anything before it.
            mutate(service, "trial", "other", &collection("other"))?;
            mutate(service, "main", "two", &collection("two"))?;
            assert_eq!(
                frames.recv_timeout(WAIT)?,
                frame("main", head(service, "main")?)
            );
            Ok(())
        })?;
        outcome?;
        assert_eq!(ended.code, "pass_expired");
        assert!(after.is_empty(), "frames after the publishes: {after:?}");
        Ok(())
    })
}

/// Fails while a follow outlives its pass, or sends a head after the expiry:
/// the wake after the pass's expiry ends it `pass_expired` and sends nothing,
/// though the branch it follows moved.
#[test]
fn a_follow_ends_pass_expired_at_the_passs_expiry() -> TestResult {
    with_service(|service| {
        let (outcome, ended, after) = with_follower(service, "main", |frames| {
            frames.recv_timeout(WAIT)?;
            Ok(())
        })?;
        outcome?;
        assert_eq!(ended.code, "pass_expired");
        assert!(after.is_empty(), "a head past the expiry: {after:?}");
        Ok(())
    })
}

/// A pass already expired when the follow starts is sent no head at all.
#[test]
fn an_expired_pass_is_sent_no_head() -> TestResult {
    with_service(|service| {
        let mut sent = Vec::new();
        let ended = service.follow(
            "main",
            Some(EXPIRES),
            || at(EXPIRES + 1),
            |head: &Value| {
                sent.push(head.clone());
                Ok(())
            },
        )?;
        assert_eq!(ended.code, "pass_expired");
        assert!(sent.is_empty(), "an expired pass was sent {sent:?}");
        Ok(())
    })
}

/// Fails while a follower of a discarded branch keeps waiting: the removal
/// wakes it and it ends `branch_unknown`; a branch never published ends it
/// the same way before any frame.
#[test]
fn a_follow_ends_branch_unknown_when_its_branch_is_removed() -> TestResult {
    with_service(|service| {
        fork_trial(service)?;
        let (outcome, ended, after) = with_follower(service, "trial", |frames| {
            assert_eq!(
                frames.recv_timeout(WAIT)?,
                frame("trial", head(service, "trial")?)
            );
            mutate(
                service,
                "trial",
                "discard",
                &json!({"kind": "branch.discard", "value": {"name": "trial"}}),
            )?;
            Ok(())
        })?;
        outcome?;
        assert_eq!(ended.code, "branch_unknown");
        assert!(after.is_empty(), "frames after the discard: {after:?}");
        let mut sent = Vec::new();
        let never = service.follow("absent", None, SystemTime::now, |head: &Value| {
            sent.push(head.clone());
            Ok(())
        })?;
        assert_eq!(never.code, "branch_unknown");
        assert!(sent.is_empty());
        Ok(())
    })
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    std::time::Duration::from_secs(1)
}