lean_lmdb

use super::*;

#[test]
fn crud_error_mapping_is_stable() {
    for (error, expected) in [
        (MdbError::MapFull, CrudError::MapFull),
        (MdbError::MapResized, CrudError::MapResized),
        (MdbError::ReadersFull, CrudError::ReadersFull),
        (MdbError::TxnFull, CrudError::TransactionFull),
        (MdbError::BadDbi, CrudError::InvalidDatabase),
        (MdbError::Incompatible, CrudError::InvalidDatabase),
        (MdbError::BadValSize, CrudError::InvalidKey),
        (MdbError::Corrupted, CrudError::Database),
    ] {
        assert_eq!(map_heed_error(heed::Error::Mdb(error)), expected);
    }
    assert_eq!(
        map_heed_error(heed::Error::Io(std::io::Error::from(
            std::io::ErrorKind::OutOfMemory
        ))),
        CrudError::OutOfMemory
    );
    assert_eq!(
        map_heed_error(heed::Error::Io(std::io::Error::from(
            std::io::ErrorKind::PermissionDenied
        ))),
        CrudError::Io
    );
}

#[test]
fn binary_crud_preserves_bytes_isolates_databases_and_is_idempotent() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("binary_crud");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let state =
        open_shared_environment(canonical.clone(), options(), true).expect("environment open");
    let first = database_resource(&state, "first");
    let second = database_resource(&state, "second");
    let key = [0, 0xff, 1];

    assert_eq!(get_binary(&first, &key), Ok(None));
    put_binary(&first, &key, &[]).expect("put empty value");
    assert_eq!(get_binary(&first, &key), Ok(Some(Vec::new())));
    assert_eq!(get_binary(&second, &key), Ok(None));
    put_binary(&first, &key, &[0xff, 0, 0x80]).expect("overwrite value");
    assert_eq!(get_binary(&first, &key), Ok(Some(vec![0xff, 0, 0x80])));
    delete_binary(&first, &key).expect("delete present key");
    delete_binary(&first, &key).expect("delete missing key");
    assert_eq!(get_binary(&first, &key), Ok(None));
    assert_eq!(
        put_binary(&first, &[], b"value"),
        Err(CrudError::InvalidKey)
    );
    assert_eq!(
        put_binary(&first, &vec![1; state.env.max_key_size() + 1], b"value"),
        Err(CrudError::InvalidKey)
    );

    drop(first);
    drop(second);
    drop(state);
    cleanup_environment(&canonical);
}

#[test]
fn clearing_stale_readers_is_safe_with_a_live_reader() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("clear_stale_readers");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let state =
        open_shared_environment(canonical.clone(), options(), true).expect("environment open");
    let transaction = state.env.read_txn().expect("read transaction");

    assert_eq!(clear_stale_readers(&state), Ok(0));
    transaction.commit().expect("commit read transaction");

    drop(state);
    cleanup_environment(&canonical);
}

#[test]
fn map_full_write_rolls_back_and_prior_value_survives() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("map_full");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let small = EnvironmentOptions {
        map_size: MIN_MAP_SIZE as usize,
        ..options()
    };
    let state = open_shared_environment(canonical.clone(), small, true).expect("environment open");
    let database = database_resource(&state, "data");
    put_binary(&database, b"key", b"before").expect("initial put");
    assert_eq!(
        put_binary(&database, b"key", &vec![7; 2 * MIN_MAP_SIZE as usize]),
        Err(CrudError::MapFull)
    );
    assert_eq!(get_binary(&database, b"key"), Ok(Some(b"before".to_vec())));

    drop(database);
    drop(state);
    cleanup_environment(&canonical);
}

#[test]
fn concurrent_crud_persists_across_reopen() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("crud_reopen");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let state =
        open_shared_environment(canonical.clone(), options(), true).expect("environment open");
    let database = Arc::new(database_resource(&state, "data"));
    let mut workers = Vec::new();
    for index in 0..16_u8 {
        let database = Arc::clone(&database);
        workers.push(thread::spawn(move || {
            put_binary(&database, &[index], &[0xff, index]).expect("concurrent put");
        }));
    }
    for worker in workers {
        worker.join().expect("worker did not panic");
    }
    drop(database);
    drop(state);
    registry().remove(&canonical);
    if let Some(closing) = heed::env_closing_event(&canonical) {
        closing.wait();
    }

    let reopened =
        open_shared_environment(canonical.clone(), options(), false).expect("reopen environment");
    let database = DatabaseResource {
        state: Arc::clone(&reopened),
        database: open_named_database(&reopened, "data").expect("reopen database"),
        name: "data".to_owned(),
    };
    for index in 0..16_u8 {
        assert_eq!(get_binary(&database, &[index]), Ok(Some(vec![0xff, index])));
    }
    drop(database);
    drop(reopened);
    cleanup_environment(&canonical);
}