haematite 0.7.0

//! A4 e2e: the cross-process data-dir writer lock and the blessed read-only
//! observer.
//!
//! Advisory file locks are per OPEN FILE DESCRIPTION, not per process, so a
//! second `Database` handle inside this one test process contends on exactly
//! the same OS lock a second process would — these tests therefore exercise
//! the real cross-writer exclusion without spawning processes.

use std::error::Error;
use std::io;
use std::path::Path;

use haematite::{Database, DatabaseConfig, DatabaseError, ReadOnlyDatabase};

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

fn config_for(path: &Path) -> DatabaseConfig {
    DatabaseConfig {
        data_dir: path.to_path_buf(),
        shard_count: 2,
        distributed: None,
        executor_threads: None,
    }
}

fn boxed_error(message: &'static str) -> Box<dyn Error> {
    Box::new(io::Error::other(message))
}

fn require_open_failure(
    result: Result<Database, DatabaseError>,
    context: &'static str,
) -> Result<DatabaseError, Box<dyn Error>> {
    match result {
        Ok(_database) => Err(boxed_error(context)),
        Err(error) => Ok(error),
    }
}

/// THE LOUD FAILURE: while one writer is live, a second `Database::open` of the
/// same data dir must fail immediately with the distinct
/// `DatabaseError::DataDirLocked` — it must not hang, and it must not boot a
/// second set of shard actors over the same WALs.
#[test]
fn second_writer_open_fails_loudly_with_data_dir_locked() -> TestResult {
    let dir = tempfile::tempdir()?;
    let writer = Database::create(config_for(dir.path()))?;

    let error = require_open_failure(
        Database::open(dir.path()),
        "a second writer open over a live writer must fail",
    )?;
    assert!(
        matches!(error, DatabaseError::DataDirLocked { .. }),
        "expected DataDirLocked, got: {error}"
    );
    assert!(
        error.to_string().contains("writer.lock"),
        "the loud failure must name the lockfile: {error}"
    );

    // The first writer is unharmed by the failed second open.
    writer.put(b"still-mine".to_vec(), b"value".to_vec())?;
    writer.commit()?;
    assert_eq!(writer.get(b"still-mine")?, Some(b"value".to_vec()));
    Ok(())
}

/// `Database::create` over a live writer's dir must fail with the SAME distinct
/// error, and must not clobber the live writer's `config.json` (the lock is
/// taken before the config write) nor clean the directory away.
#[test]
fn create_over_a_live_writer_fails_and_leaves_the_dir_intact() -> TestResult {
    let dir = tempfile::tempdir()?;
    let writer = Database::create(config_for(dir.path()))?;
    let config_before = std::fs::read(dir.path().join("config.json"))?;

    let mut clobbering = config_for(dir.path());
    clobbering.shard_count = 7;
    let error = require_open_failure(
        Database::create(clobbering),
        "a second create over a live writer must fail",
    )?;
    assert!(
        matches!(error, DatabaseError::DataDirLocked { .. }),
        "expected DataDirLocked, got: {error}"
    );

    // Loser must not have rewritten the config or removed the winner's dir.
    let config_after = std::fs::read(dir.path().join("config.json"))?;
    assert_eq!(
        config_before, config_after,
        "a losing create must not clobber the live writer's config"
    );
    drop(writer);
    Ok(())
}

/// Advisory locks release with the handle: dropping the writer lets the same
/// process (or any other) reopen the dir, repeatedly.
#[test]
fn dropping_the_writer_releases_the_lock_for_reopen() -> TestResult {
    let dir = tempfile::tempdir()?;
    let first = Database::create(config_for(dir.path()))?;
    first.put(b"persisted".to_vec(), b"across-reopen".to_vec())?;
    first.commit()?;
    drop(first);

    let second = Database::open(dir.path())?;
    assert_eq!(second.get(b"persisted")?, Some(b"across-reopen".to_vec()));
    drop(second);

    // Release is not a one-shot: a third open succeeds too.
    let third = Database::open(dir.path())?;
    assert_eq!(third.get(b"persisted")?, Some(b"across-reopen".to_vec()));
    Ok(())
}

/// THE BLESSING: a read-only observer opens alongside a LIVE writer (no lock
/// contention), sees exactly the committed state — never the writer's
/// uncommitted WAL buffer — and tracks each new commit.
#[test]
fn read_only_observer_succeeds_alongside_a_live_writer() -> TestResult {
    let dir = tempfile::tempdir()?;
    let writer = Database::create(config_for(dir.path()))?;
    writer.put(b"committed-key".to_vec(), b"committed-value".to_vec())?;
    writer.commit()?;
    writer.put(b"buffered-key".to_vec(), b"uncommitted".to_vec())?;

    // Observer opens while the writer holds the exclusive lock.
    let observer = ReadOnlyDatabase::open(dir.path())?;
    assert_eq!(observer.shard_count(), 2);
    assert_eq!(
        observer.get(b"committed-key")?,
        Some(b"committed-value".to_vec())
    );
    // Committed state ONLY: the writer's WAL buffer is invisible.
    assert_eq!(
        observer.get(b"buffered-key")?,
        None,
        "an observer must never see uncommitted buffered writes"
    );

    // The observer tracks the writer's next commit.
    writer.commit()?;
    assert_eq!(
        observer.get(b"buffered-key")?,
        Some(b"uncommitted".to_vec())
    );

    // And it holds NO lock: with the observer still live, the writer seat can
    // be handed over to a new writer.
    drop(writer);
    let successor = Database::open(dir.path())?;
    assert_eq!(
        observer.get(b"committed-key")?,
        Some(b"committed-value".to_vec()),
        "the observer keeps reading across a writer handover"
    );
    drop(successor);
    Ok(())
}

/// Observer `range` mirrors the live shard-local range semantics over committed
/// state, and a committed tombstone (delete) reads as absent.
#[test]
fn observer_range_reads_committed_entries_and_hides_tombstones() -> TestResult {
    let dir = tempfile::tempdir()?;
    let writer = Database::create(config_for(dir.path()))?;

    // Keys co-located on the shard owning `range:` so one shard-local range
    // sees them all (mirrors Database::range routing by `from`).
    let target_shard = writer.shard_for(b"range:");
    let mut keys: Vec<Vec<u8>> = Vec::new();
    let mut candidate = 0_u64;
    while keys.len() < 3 {
        let key = format!("range:{candidate:04}").into_bytes();
        if writer.shard_for(&key) == target_shard {
            keys.push(key);
        }
        candidate = candidate.saturating_add(1);
        assert!(candidate < 10_000, "failed to find enough shard-local keys");
    }
    for key in &keys {
        writer.put(key.clone(), b"live".to_vec())?;
    }
    writer.commit()?;
    // Delete the middle key: a stamped tombstone, committed.
    writer.delete(keys[1].clone())?;
    writer.commit()?;

    let observer = ReadOnlyDatabase::open(dir.path())?;
    let entries = observer.range(b"range:", b"range:\xff")?;
    let expected: Vec<(Vec<u8>, Vec<u8>)> = vec![
        (keys[0].clone(), b"live".to_vec()),
        (keys[2].clone(), b"live".to_vec()),
    ];
    assert_eq!(
        entries, expected,
        "a committed tombstone must read as absent to the observer"
    );
    assert_eq!(observer.get(&keys[1])?, None);

    // Empty and inverted ranges return empty without touching a shard.
    assert!(observer.range(b"zz", b"aa")?.is_empty());
    drop(writer);
    Ok(())
}

/// A never-committed database (and a never-materialised shard) reads as empty;
/// the committed-root projection reports `None`, and an out-of-range shard id
/// is rejected.
#[test]
fn observer_reads_uncommitted_and_unmaterialised_shards_as_empty() -> TestResult {
    let dir = tempfile::tempdir()?;
    let writer = Database::create(config_for(dir.path()))?;

    let observer = ReadOnlyDatabase::open(dir.path())?;
    assert_eq!(observer.get(b"anything")?, None);
    assert!(observer.range(b"a", b"z")?.is_empty());
    assert_eq!(observer.committed_root(0)?, None);
    assert_eq!(observer.committed_root(1)?, None);
    assert!(
        observer.committed_root(2).is_err(),
        "shard id at shard_count must be rejected"
    );

    // After one commit the touched shard exposes a committed root.
    writer.put(b"first".to_vec(), b"commit".to_vec())?;
    writer.commit()?;
    let shard = observer.shard_for(b"first");
    assert!(
        observer.committed_root(shard)?.is_some(),
        "a committed shard must expose its committed root"
    );
    Ok(())
}

/// Opening an observer over a directory that is not a database fails with the
/// observer's config error, not a panic or a silent empty database.
#[test]
fn observer_open_on_a_non_database_dir_fails() -> TestResult {
    let dir = tempfile::tempdir()?;
    match ReadOnlyDatabase::open(dir.path()) {
        Ok(_observer) => Err(boxed_error(
            "opening an observer over a non-database dir must fail",
        )),
        Err(error) => {
            assert!(!error.to_string().is_empty());
            Ok(())
        }
    }
}