haematite-cli 0.21.8

//! EMB-E2: a record's windows committed into the companion, (b), (d), (e)
//! and (f).

use std::error::Error;
use std::num::NonZeroUsize;

use super::{
    FIELD, TestResult, code, companion, create, declare, head, made, mutate, page, position, put,
    records, revision, stored,
};
use crate::serve::protocol::{Change, DocumentDelete, Fork, Version};
use crate::serve::storage::embed_inherit::Inherited;
use crate::serve::storage::embed_job::{MadeWindows, Page, Window};
use crate::serve::storage::embedded;
use crate::serve::storage::{Failure, Storage, branch_name, reads};

/// `id`'s windows over `spans`, made from `made_from`.
fn windows(id: &str, made_from: Version, spans: &[(u64, u64)]) -> MadeWindows {
    MadeWindows {
        id: id.to_owned(),
        made_from,
        windows: spans
            .iter()
            .map(|&(start, end)| Window {
                start,
                end,
                vector: vec![1.0, 0.0],
            })
            .collect(),
    }
}

/// A page that ends its walk toward `toward` with `windowed` records.
fn windowed(windowed: Vec<MadeWindows>, toward: Version) -> Page {
    Page {
        windowed,
        ..page(Vec::new(), &[], toward)
    }
}

/// Window `index` of `id` in the companion, if it is there.
fn window(storage: &Storage, id: &str, index: u32) -> Result<Option<Vec<u8>>, Box<dyn Error>> {
    let at = companion(storage).ok_or("no companion")?;
    let key = embedded::window_key("notes", FIELD, id, index);
    Ok(reads::get_bytes(&storage.reader, &at, &key)?)
}

/// `n-1` written, the field declared, and `n-1` embedded as three windows.
fn three(storage: &mut Storage) -> TestResult {
    put(storage, "n-1", "a long text")?;
    declare(storage)?;
    let toward = head(storage)?;
    let long = windows(
        "n-1",
        revision(storage, "n-1")?,
        &[(0, 6), (4, 10), (8, 11)],
    );
    storage.embed_commit("main", &windowed(vec![long], toward), records()?)?;
    Ok(())
}

/// Fails while spans with a gap are stored: the page is refused
/// `embed_windows_invalid` and commits nothing.
#[test]
fn spans_with_a_gap_fail_the_page_by_name() -> TestResult {
    let (_directory, mut storage) = create()?;
    put(&mut storage, "n-1", "a long text")?;
    declare(&mut storage)?;
    let toward = head(&storage)?;
    let gapped = windows("n-1", revision(&storage, "n-1")?, &[(0, 10), (12, 20)]);
    let refused = storage.embed_commit("main", &windowed(vec![gapped], toward), records()?);
    assert_eq!(code(refused)?, "embed_windows_invalid");
    assert_eq!(position(&storage)?, None);
    assert!(companion(&storage).is_none(), "nothing is committed");
    Ok(())
}

/// Fails while a window that starts with the one before it, or ends with
/// it, is stored, or while one that starts where the one before it ends is
/// refused.
#[test]
fn window_order_boundaries_are_exact() -> TestResult {
    for (spans, refused) in [
        (&[(0, 6), (0, 11)][..], true),
        (&[(0, 11), (5, 11)][..], true),
        (&[(0, 6), (6, 11)][..], false),
    ] {
        let (_directory, mut storage) = create()?;
        put(&mut storage, "n-1", "a long text")?;
        declare(&mut storage)?;
        let toward = head(&storage)?;
        let made = windows("n-1", revision(&storage, "n-1")?, spans);
        let result = storage.embed_commit("main", &windowed(vec![made], toward), records()?);
        if refused {
            assert_eq!(code(result)?, "embed_windows_invalid");
        } else {
            result?;
        }
    }
    Ok(())
}

/// Fails while a record is stored as one window: fewer than two is not a
/// cover, and is refused by name.
#[test]
fn one_window_is_refused_by_name() -> TestResult {
    let (_directory, mut storage) = create()?;
    put(&mut storage, "n-1", "text")?;
    declare(&mut storage)?;
    let toward = head(&storage)?;
    let one = windows("n-1", revision(&storage, "n-1")?, &[(0, 4)]);
    let refused = storage.embed_commit("main", &windowed(vec![one], toward), records()?);
    assert_eq!(code(refused)?, "embed_windows_invalid");
    Ok(())
}

/// Fails while a record with more windows than [`embedded::WINDOWS_MOST`]
/// is stored, each window in order over the one before it.
#[test]
fn windows_over_the_most_are_refused_by_name() -> TestResult {
    let (_directory, mut storage) = create()?;
    put(&mut storage, "n-1", "a long text")?;
    declare(&mut storage)?;
    let toward = head(&storage)?;
    let spans: Vec<(u64, u64)> = (0..=u64::from(embedded::WINDOWS_MOST))
        .map(|start| (start, start + 2))
        .collect();
    let over = windows("n-1", revision(&storage, "n-1")?, &spans);
    let refused = storage.embed_commit("main", &windowed(vec![over], toward), records()?);
    assert_eq!(code(refused)?, "embed_windows_invalid");
    Ok(())
}

/// Fails while the windows are not written whole under `vw\0` with their
/// spans and count, or while a `v\0` vector is written for them.
#[test]
fn a_record_s_windows_are_written_whole_with_their_spans() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    let mut spans = Vec::new();
    for index in 0..3 {
        let bytes = window(&storage, "n-1", index)?.ok_or("a window is missing")?;
        let (span, _provenance, vector) = embedded::decode_window(&bytes, 2)?;
        spans.push((span.count, span.start, span.end));
        assert_eq!(vector, [1.0, 0.0]);
    }
    assert_eq!(spans, [(3, 0, 6), (3, 4, 10), (3, 8, 11)]);
    assert_eq!(window(&storage, "n-1", 3)?, None);
    assert_eq!(stored(&storage, "n-1")?, None);
    Ok(())
}

/// Fails while a stale window is left: a record embedded as three windows,
/// then made short, keeps one `v\0` vector and no window, cleared in the
/// commit that writes it.
#[test]
fn a_windowed_text_made_short_keeps_one_vector() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    put(&mut storage, "n-1", "short")?;
    let toward = head(&storage)?;
    let short = made("n-1", revision(&storage, "n-1")?, &[0.0, 1.0]);
    let committed = storage.embed_commit("main", &page(vec![short], &[], toward), records()?)?;
    assert_eq!(committed.written, 1, "the clearing deletes are not counted");
    for index in 0..3 {
        assert_eq!(window(&storage, "n-1", index)?, None, "window {index}");
    }
    assert!(stored(&storage, "n-1")?.is_some());
    Ok(())
}

/// Fails while a short text made long keeps its one vector beside its
/// windows.
#[test]
fn a_short_text_made_long_loses_its_one_vector() -> TestResult {
    let (_directory, mut storage) = create()?;
    put(&mut storage, "n-1", "short")?;
    declare(&mut storage)?;
    let first = head(&storage)?;
    let short = made("n-1", revision(&storage, "n-1")?, &[0.0, 1.0]);
    storage.embed_commit("main", &page(vec![short], &[], first), records()?)?;
    put(&mut storage, "n-1", "a long text")?;
    let second = head(&storage)?;
    let long = windows("n-1", revision(&storage, "n-1")?, &[(0, 6), (4, 11)]);
    storage.embed_commit("main", &windowed(vec![long], second), records()?)?;
    assert_eq!(stored(&storage, "n-1")?, None);
    assert!(window(&storage, "n-1", 1)?.is_some());
    Ok(())
}

/// Fails while windows past a new, shorter cover are left: three windows
/// then two leave window 2 gone and window 0 naming two.
#[test]
fn a_rewritten_text_with_fewer_windows_clears_the_rest() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    put(&mut storage, "n-1", "a long tex")?;
    let toward = head(&storage)?;
    let fewer = windows("n-1", revision(&storage, "n-1")?, &[(0, 6), (4, 10)]);
    storage.embed_commit("main", &windowed(vec![fewer], toward), records()?)?;
    assert_eq!(window(&storage, "n-1", 2)?, None);
    let first = window(&storage, "n-1", 0)?.ok_or("no window 0")?;
    assert_eq!(embedded::window_span(&first)?.count, 2);
    Ok(())
}

/// (e): fails while a record past the cap keeps its old windows: one the
/// page clears loses every window.
#[test]
fn a_cleared_record_loses_its_windows() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    put(&mut storage, "n-1", "a far longer text")?;
    let toward = head(&storage)?;
    let cleared = Page {
        cleared: vec!["n-1".to_owned()],
        ..page(Vec::new(), &[], toward)
    };
    storage.embed_commit("main", &cleared, records()?)?;
    for index in 0..3 {
        assert_eq!(window(&storage, "n-1", index)?, None, "window {index}");
    }
    Ok(())
}

/// Fails while a removed record keeps its windows.
#[test]
fn a_removed_record_loses_its_windows() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    let delete = DocumentDelete {
        collection: "notes".to_owned(),
        document_id: "n-1".to_owned(),
    };
    mutate(&mut storage, Change::Delete(delete))?;
    let toward = head(&storage)?;
    storage.embed_commit("main", &page(Vec::new(), &["n-1"], toward), records()?)?;
    for index in 0..3 {
        assert_eq!(window(&storage, "n-1", index)?, None, "window {index}");
    }
    Ok(())
}

/// The `vw\0` keys `name` holds, at most `most`.
fn window_keys(storage: &Storage, name: &str, most: usize) -> Result<usize, Box<dyn Error>> {
    let record = storage.refs.get(name).ok_or("no such record")?;
    let most = NonZeroUsize::new(most).ok_or("most is zero")?;
    let mut found = 0;
    for shard in &record.shards {
        let page = storage
            .reader
            .range_page_at(shard.shard_id, shard.head, b"vw\0", b"vw\x01", most)
            .map_err(Failure::observer)?;
        found += page.entries.len();
    }
    Ok(found)
}

/// (f): `merge_coverage.rs:312` and `:637`, `merge_candidate.rs:359`,
/// `admission_reads.rs:289` and `vectors.rs` read the branch only, never
/// its companion. Fails while a window is written anywhere but the
/// companion.
#[test]
fn windows_are_written_to_the_companion_only() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    assert_eq!(window_keys(&storage, &branch_name("main"), 8)?, 0);
    let name = embedded::companion_name("main");
    assert_eq!(window_keys(&storage, &name, 8)?, 3);
    Ok(())
}

/// (f), `embed_inherit`: fails while a fork's companion is made without its
/// parent's windows. It shares the parent's nodes and moves positions only,
/// so the windows come with it unread.
#[test]
fn a_fork_inherits_its_parent_s_windows() -> TestResult {
    let (_directory, mut storage) = create()?;
    three(&mut storage)?;
    let from = head(&storage)?;
    let fork = Fork {
        name: "feature".to_owned(),
        from,
    };
    mutate(&mut storage, Change::Fork(fork))?;
    assert_eq!(storage.embed_inherit("feature")?, Inherited::Done);
    let name = embedded::companion_name("feature");
    assert_eq!(window_keys(&storage, &name, 8)?, 3);
    Ok(())
}