haematite-cli 0.21.7

//! STOR-003 H1c: reads of a service store that reclaims, with a pass placed
//! while a read is under way.
//!
//! A read runs on its own thread. Once its capture is held (the test seam
//! `after_held`), it waits while this thread moves `main` on and runs every
//! shard's reclaim to rest, then finishes. Every wait is bounded and ended by
//! a signal; nothing sleeps.

use std::collections::BTreeMap;
use std::error::Error;
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::path::PathBuf;
use std::sync::mpsc;
use std::time::Duration;

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

use super::{ChangesRequest, Context as ChangesContext, Moment, Settings as ChangesSettings, list};
use crate::serve::access::{self, Standalone};
use crate::serve::protocol::{
    Change, CollectionWrite, DocumentWrite, Mutation, QueryPage, QueryRequest, Selection, Version,
};
use crate::serve::query::admission::Slots;
use crate::serve::query::run::{self, Context as QueryContext};
use crate::serve::query::{Rows, Seal, Settings as QuerySettings};
use crate::serve::storage::{Catalogue, Failure, Storage, after_held};

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

/// The longest any signal here is waited for; a red ends the wait in error.
const BOUND: Duration = Duration::from_secs(60);

struct Fixture {
    directory: tempfile::TempDir,
    storage: Storage,
    reader: ReadOnlyDatabase,
    catalogue: Catalogue,
}

fn head() -> Selection {
    Selection {
        branch: "main".to_owned(),
        version: None,
    }
}

/// `change` written on main at its head; the version it made.
fn write(
    storage: &mut Storage,
    request_id: &str,
    change: Change,
) -> Result<Version, Box<dyn Error>> {
    let mutation = Mutation {
        request_id: request_id.to_owned(),
        branch: "main".to_owned(),
        expected: storage.capture(&head())?.version,
        change,
    };
    Ok(storage.mutate(&mutation)?.version)
}

fn put(id: &str) -> Result<Change, Box<dyn Error>> {
    Ok(Change::Put(DocumentWrite {
        collection: "records".to_owned(),
        document_id: id.to_owned(),
        document: RawValue::from_string(format!(r#"{{"id":"{id}"}}"#))?,
    }))
}

/// A two-shard store created as the service creates one, under reclaim, with
/// `ids` written in `records`.
fn fixture(ids: &[&str]) -> Result<Fixture, Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 2,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let declare = Change::Collection(CollectionWrite {
        collection: "records".to_owned(),
        schema: None,
    });
    write(&mut storage, "declare", declare)?;
    for id in ids {
        write(&mut storage, id, put(id)?)?;
    }
    let reader = storage.reader()?;
    let catalogue = storage.catalogue();
    Ok(Fixture {
        directory,
        storage,
        reader,
        catalogue,
    })
}

/// Run `read` on its own thread; at its capture held after the first `skip`,
/// give `between` the storage, then let the read finish.
fn across_a_pass<T: Send>(
    storage: &mut Storage,
    skip: usize,
    read: impl FnOnce() -> Result<T, Failure> + Send,
    between: impl FnOnce(&mut Storage) -> TestResult,
) -> Result<Result<T, Failure>, Box<dyn Error>> {
    let (held, reached) = mpsc::channel::<()>();
    let (resume, resumed) = mpsc::channel::<()>();
    std::thread::scope(|scope| -> Result<Result<T, Failure>, Box<dyn Error>> {
        let reading = scope.spawn(move || {
            after_held::set(skip, move || {
                if held.send(()).is_ok() {
                    // Ended by the go-ahead, or by the bound once the test has failed.
                    let _resumed = resumed.recv_timeout(BOUND);
                }
            });
            read()
        });
        let moved = reached
            .recv_timeout(BOUND)
            .map_err(|_timeout| Box::<dyn Error>::from("the read held no capture"))
            .and_then(|()| between(storage));
        // The read may have ended already, never having been held.
        let _told = resume.send(());
        let answer = reading
            .join()
            .map_err(|_panic| "the reading thread panicked")?;
        moved.map(|()| answer)
    })
}

/// `main` moves on by one write, then every shard's reclaim runs to rest.
fn move_on_and_pass(storage: &mut Storage) -> TestResult {
    write(storage, "later", put("later")?)?;
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    Ok(())
}

fn query_settings() -> QuerySettings {
    QuerySettings {
        time_ms: NonZeroU64::MIN.saturating_add(999),
        memory_bytes: NonZeroU64::MIN.saturating_add((1 << 26) - 1),
        outputs_per_record: NonZeroU64::MIN,
        scan_records: NonZeroU64::MIN.saturating_add(98),
        max_running: NonZeroUsize::MIN,
        worker: PathBuf::from("/no/such/haematite-jq"),
    }
}

/// A page of the origins of `.` (no worker), `rows` at a time, after
/// `previous` if given.
fn origins(previous: Option<&Value>, rows: u64) -> Result<QueryRequest, Box<dyn Error>> {
    let mut selection = head();
    let continuation = if let Some(answer) = previous {
        selection.version = Some(serde_json::from_value(
            answer["selection"]["version"].clone(),
        )?);
        Some(answer["next"].as_str().ok_or("no next")?.to_owned())
    } else {
        None
    };
    Ok(QueryRequest {
        selection,
        query: ".".to_owned(),
        parameters: BTreeMap::new(),
        page: QueryPage {
            rows: NonZeroU64::new(rows).ok_or("zero rows")?,
            bytes: NonZeroU64::MIN.saturating_add((1 << 20) - 1),
        },
        continuation,
        rows: Rows::Origins,
    })
}

fn query(
    reader: &ReadOnlyDatabase,
    catalogue: &Catalogue,
    seal: &Seal,
    request: &QueryRequest,
) -> Result<Value, Failure> {
    let access = access::request(&Standalone)?;
    let settings = query_settings();
    let slots = Slots::new(NonZeroUsize::MIN);
    let context = QueryContext {
        access: &access,
        reader,
        catalogue,
        slots: &slots,
        settings: Some(&settings),
        max_page: NonZeroU32::MIN.saturating_add(99),
        seal,
    };
    let text = run::run(&context, request, None)?;
    serde_json::from_str(text.get()).map_err(Failure::storage)
}

/// Each row's origin, as `collection/id`.
fn given(answer: &Value) -> Result<Vec<String>, Box<dyn Error>> {
    let rows = answer["rows"].as_array().ok_or("no rows array")?;
    rows.iter()
        .map(|row| {
            let origin = &row["origin"];
            let collection = origin["collection"].as_str().ok_or("no collection")?;
            let id = origin["id"].as_str().ok_or("no id")?;
            Ok(format!("{collection}/{id}"))
        })
        .collect()
}

/// Fails while a page's capture is not held for the page: a query page is
/// captured, `main` moves on and a full pass runs, and the page's answer is
/// whole and names the version it read. No hold lives after it.
///
/// A control, not the proof: history rows keep every version, so the seam
/// shows the read is held across the pass, not that the hold alone keeps its
/// roots. That proof is the engine's (R4) until a version can be released.
#[test]
fn a_page_already_begun_finishes_on_the_roots_it_captured() -> TestResult {
    let mut fixture = fixture(&["r1", "r2", "r3"])?;
    let read_at = fixture.storage.capture(&head())?.version;
    let seal = Seal::new()?;
    let request = origins(None, 100)?;
    let (reader, catalogue) = (&fixture.reader, &fixture.catalogue);
    let answer = across_a_pass(
        &mut fixture.storage,
        0,
        move || query(reader, catalogue, &seal, &request),
        move_on_and_pass,
    )??;
    assert_eq!(given(&answer)?, ["records/r1", "records/r2", "records/r3"]);
    assert_eq!(answer["selection"]["version"], json!(read_at));
    assert_eq!(fixture.storage.live_read_holds(), 0);
    Ok(())
}

/// Fails while a continuation is answered from another version than its own:
/// with every version kept by its history row, the next page after a write
/// and a full pass is answered at the first page's version.
#[test]
fn the_next_page_of_a_kept_version_is_answered_after_a_pass() -> TestResult {
    let mut fixture = fixture(&["r1", "r2", "r3"])?;
    let seal = Seal::new()?;
    let first = query(
        &fixture.reader,
        &fixture.catalogue,
        &seal,
        &origins(None, 2)?,
    )?;
    assert_eq!(fixture.storage.live_read_holds(), 0);
    move_on_and_pass(&mut fixture.storage)?;
    let second = query(
        &fixture.reader,
        &fixture.catalogue,
        &seal,
        &origins(Some(&first), 2)?,
    )?;
    assert_eq!(given(&second)?, ["records/r3"]);
    assert_eq!(
        second["selection"]["version"],
        first["selection"]["version"]
    );
    assert_eq!(fixture.storage.live_read_holds(), 0);
    Ok(())
}

/// Fails while a version whose roots have gone is read anyway or answered
/// from a newer one: once the first page's version has gone, the next page is
/// refused naming it. No hold lives after the refusal.
///
/// A control, not the proof: nothing releases a version here, so its roots
/// are taken by hand. The store keeps no record of a release for it
/// (`history::read_released` does not cover it; the capture above succeeds),
/// so the true answer is `version_missing`: `version_released` is said only
/// on the store's own record of a release (RX2, 6B; catalogue.rs).
#[test]
fn the_next_page_of_a_released_version_is_refused_by_name() -> TestResult {
    let mut fixture = fixture(&["r1", "r2", "r3"])?;
    let seal = Seal::new()?;
    let first = query(
        &fixture.reader,
        &fixture.catalogue,
        &seal,
        &origins(None, 2)?,
    )?;
    let version: Version = serde_json::from_value(first["selection"]["version"].clone())?;
    move_on_and_pass(&mut fixture.storage)?;
    let released = fixture.storage.capture(&Selection {
        branch: "main".to_owned(),
        version: Some(version),
    })?;
    let mut taken = 0_usize;
    for shard in &released.shards {
        let hex = shard.head.to_string();
        let (prefix, name) = hex.split_at(2);
        let store = fixture
            .directory
            .path()
            .join("database")
            .join(format!("shard-{}", shard.shard_id))
            .join("store");
        match std::fs::remove_file(store.join(prefix).join(name)) {
            Ok(()) => taken += 1,
            // An empty shard's head is no file.
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
    }
    assert!(taken > 0, "no root of the version was taken");
    let refused = query(
        &fixture.reader,
        &fixture.catalogue,
        &seal,
        &origins(Some(&first), 2)?,
    );
    let failure = refused
        .err()
        .ok_or("a page was answered from a released version")?;
    assert_eq!(failure.code, "version_missing", "{}", failure.message);
    let named = format!("version {}.{}", version.generation, version.sequence);
    assert!(failure.message.contains(&named), "{}", failure.message);
    assert_eq!(fixture.storage.live_read_holds(), 0);
    Ok(())
}

fn at(version: Version) -> Moment {
    Moment::At(Selection {
        branch: "main".to_owned(),
        version: Some(version),
    })
}

/// Fails while a changes page's second moment is not held: between two kept
/// versions, with a full pass run after both are captured, the answer is
/// whole. No hold lives after it.
#[test]
fn a_changes_page_is_whole_after_a_pass_between_capture_and_answer() -> TestResult {
    let mut fixture = fixture(&["r1"])?;
    let from = fixture.storage.capture(&head())?.version;
    let to = write(&mut fixture.storage, "r2", put("r2")?)?;
    let request = ChangesRequest {
        from: at(from),
        to: at(to),
        rows: Rows::Origins,
        page: QueryPage {
            rows: NonZeroU64::MIN.saturating_add(9),
            bytes: NonZeroU64::MIN.saturating_add((1 << 20) - 1),
        },
        continuation: None,
    };
    let seal = Seal::new()?;
    let settings = ChangesSettings::default();
    let (reader, catalogue) = (&fixture.reader, &fixture.catalogue);
    let answer = across_a_pass(
        &mut fixture.storage,
        1,
        || {
            let access = access::request(&Standalone)?;
            let context = ChangesContext {
                access: &access,
                reader,
                catalogue,
                settings: &settings,
                seal: &seal,
            };
            list(&context, &request, None).and_then(|text| {
                serde_json::from_str::<Value>(text.get()).map_err(Failure::storage)
            })
        },
        move_on_and_pass,
    )??;
    assert_eq!(given(&answer)?, ["records/r2"]);
    assert_eq!(answer["rows"][0]["kind"], "added");
    assert_eq!(answer["rows"][0]["after"], json!(to));
    assert_eq!(fixture.storage.live_read_holds(), 0);
    Ok(())
}

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