Current section
Files
Jump to
Current section
Files
native/lean_lmdb_nif/src/lib.rs
use heed::types::Bytes;
use heed::{Database, Env, EnvFlags, EnvOpenOptions, MdbError, WithoutTls};
use rustler::{
Atom, Binary, Decoder, Encoder, Env as NifEnv, OwnedBinary, Resource, ResourceArc, Term,
};
use std::collections::HashMap;
use std::collections::hash_map::RandomState;
use std::fs;
use std::hash::{BuildHasher, Hasher};
use std::ops::Bound;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, LazyLock, Mutex, MutexGuard, Weak};
mod atoms {
rustler::atoms! {
invalid_path,
path_not_found,
invalid_environment_options,
invalid_environment,
incompatible_environment_options,
environment_open_failed,
resource_id_exhausted,
invalid_database_name,
read_only,
transaction_failed,
database_create_failed,
database_open_failed,
database_not_found,
invalid_database,
invalid_key,
invalid_value,
map_full,
map_resized,
readers_full,
transaction_full,
out_of_memory,
io_error,
database_error,
invalid_batch,
batch_too_large,
mixed_environments,
invalid_expected,
invalid_replacement,
not_found,
missing,
put,
delete,
conflict,
ok,
prefix,
range,
nil,
done,
invalid_mode,
invalid_limit,
invalid_max_bytes,
row_too_large,
invalid_continuation,
stale_continuation,
#[cfg(all(debug_assertions, feature = "test-support"))]
test_timeout
}
}
type NifResult<T> = Result<T, rustler::Atom>;
type InternalResult<T> = Result<T, LifecycleError>;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum LifecycleError {
InvalidPath,
PathNotFound,
InvalidEnvironmentOptions,
InvalidEnvironment,
IncompatibleEnvironmentOptions,
EnvironmentOpenFailed,
ResourceIdExhausted,
InvalidDatabaseName,
ReadOnly,
TransactionFailed,
DatabaseCreateFailed,
DatabaseOpenFailed,
DatabaseNotFound,
}
impl LifecycleError {
fn atom(self) -> rustler::Atom {
match self {
Self::InvalidPath => atoms::invalid_path(),
Self::PathNotFound => atoms::path_not_found(),
Self::InvalidEnvironmentOptions => atoms::invalid_environment_options(),
Self::InvalidEnvironment => atoms::invalid_environment(),
Self::IncompatibleEnvironmentOptions => atoms::incompatible_environment_options(),
Self::EnvironmentOpenFailed => atoms::environment_open_failed(),
Self::ResourceIdExhausted => atoms::resource_id_exhausted(),
Self::InvalidDatabaseName => atoms::invalid_database_name(),
Self::ReadOnly => atoms::read_only(),
Self::TransactionFailed => atoms::transaction_failed(),
Self::DatabaseCreateFailed => atoms::database_create_failed(),
Self::DatabaseOpenFailed => atoms::database_open_failed(),
Self::DatabaseNotFound => atoms::database_not_found(),
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct EnvironmentOptions {
map_size: usize,
max_dbs: u32,
max_readers: u32,
read_only: bool,
}
type BinaryDatabase = Database<Bytes, Bytes>;
struct EnvironmentState {
id: u64,
path: PathBuf,
options: EnvironmentOptions,
env: Env<WithoutTls>,
databases: Mutex<HashMap<String, BinaryDatabase>>,
}
struct EnvironmentResource {
state: Arc<EnvironmentState>,
}
#[rustler::resource_impl]
impl Resource for EnvironmentResource {}
struct DatabaseResource {
state: Arc<EnvironmentState>,
database: BinaryDatabase,
name: String,
}
#[rustler::resource_impl]
impl Resource for DatabaseResource {}
static REGISTRY: LazyLock<Mutex<HashMap<PathBuf, Weak<EnvironmentState>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
static NEXT_RESOURCE_ID: AtomicU64 = AtomicU64::new(1);
static SCAN_TOKEN_HASHER: LazyLock<RandomState> = LazyLock::new(RandomState::new);
fn registry() -> MutexGuard<'static, HashMap<PathBuf, Weak<EnvironmentState>>> {
// A panic in unrelated native code must not permanently disable lifecycle calls.
match REGISTRY.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
const MIN_MAP_SIZE: u64 = 1024 * 1024;
const MAX_MAP_SIZE: u64 = 1024 * 1024 * 1024 * 1024;
const MAX_BATCH_OPERATIONS: usize = 10_000;
const MAX_BATCH_BYTES: usize = 64 * 1024 * 1024;
const MAX_SCAN_LIMIT: u64 = 10_000;
const MAX_SCAN_BYTES: u64 = 64 * 1024 * 1024;
const MAX_SCAN_TOKEN_BYTES: usize = 4096;
const SCAN_TOKEN_MAGIC: &[u8; 4] = b"LLS1";
fn validate_options(
map_size: u64,
max_dbs: u32,
max_readers: u32,
read_only: bool,
create: bool,
) -> InternalResult<EnvironmentOptions> {
if !(MIN_MAP_SIZE..=MAX_MAP_SIZE).contains(&map_size)
|| !map_size.is_multiple_of(page_size::get() as u64)
|| !(1..=1024).contains(&max_dbs)
|| !(1..=4096).contains(&max_readers)
|| (read_only && create)
{
return Err(LifecycleError::InvalidEnvironmentOptions);
}
let map_size =
usize::try_from(map_size).map_err(|_| LifecycleError::InvalidEnvironmentOptions)?;
Ok(EnvironmentOptions {
map_size,
max_dbs,
max_readers,
read_only,
})
}
fn canonical_environment_path(path: &str, create: bool) -> InternalResult<PathBuf> {
if path.is_empty() {
return Err(LifecycleError::InvalidPath);
}
let requested = Path::new(path);
if create {
fs::create_dir_all(requested).map_err(|_| LifecycleError::InvalidPath)?;
} else if !requested.is_dir() {
return Err(LifecycleError::PathNotFound);
}
let canonical = fs::canonicalize(requested).map_err(|_| LifecycleError::InvalidPath)?;
if canonical.to_str().is_none() {
return Err(LifecycleError::InvalidPath);
}
Ok(canonical)
}
fn next_resource_id() -> InternalResult<u64> {
NEXT_RESOURCE_ID
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| {
(id != u64::MAX).then_some(id + 1)
})
.map_err(|_| LifecycleError::ResourceIdExhausted)
}
fn open_shared_environment(
path: PathBuf,
options: EnvironmentOptions,
create: bool,
) -> InternalResult<Arc<EnvironmentState>> {
// Keep the lock across the native open. This is intentionally conservative: it
// makes lookup-and-create atomic and prevents duplicate opens for all aliases.
let mut entries = registry();
entries.retain(|_, state| state.strong_count() != 0);
// LMDB may create files during open, so enforce create=false ourselves.
if !create && !path.join("data.mdb").is_file() {
return Err(LifecycleError::PathNotFound);
}
if let Some(state) = entries.get(&path).and_then(Weak::upgrade) {
return if state.options == options {
Ok(state)
} else {
Err(LifecycleError::IncompatibleEnvironmentOptions)
};
}
// Weak::upgrade can fail while the final Arc is still running its destructor.
// heed keeps a closing event until its own process-wide entry is gone; waiting
// here prevents a transient EnvAlreadyOpened during immediate drop-and-reopen.
if let Some(closing) = heed::env_closing_event(&path) {
closing.wait();
}
let mut builder = EnvOpenOptions::new().read_txn_without_tls();
builder
.map_size(options.map_size)
.max_dbs(options.max_dbs)
.max_readers(options.max_readers);
if options.read_only {
// READ_ONLY is LMDB's safe, locking read-only mode; no unsafe durability or
// locking flags are exposed to Elixir.
unsafe {
builder.flags(EnvFlags::READ_ONLY);
}
}
// SAFETY: the canonical directory is owned externally and callers are not
// offered flags or APIs that bypass LMDB locking. Transactions never escape a
// NIF call, and this registry prevents duplicate Env values in this process.
let env = unsafe { builder.open(&path) }.map_err(|_| LifecycleError::EnvironmentOpenFailed)?;
let state = Arc::new(EnvironmentState {
id: next_resource_id()?,
path: path.clone(),
options,
env,
databases: Mutex::new(HashMap::new()),
});
entries.insert(path, Arc::downgrade(&state));
Ok(state)
}
fn validate_database_name(name: &[u8]) -> InternalResult<&str> {
if name.is_empty() || name.len() > 511 || name.contains(&0) {
return Err(LifecycleError::InvalidDatabaseName);
}
std::str::from_utf8(name).map_err(|_| LifecycleError::InvalidDatabaseName)
}
fn database_cache(state: &EnvironmentState) -> MutexGuard<'_, HashMap<String, BinaryDatabase>> {
match state.databases.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn create_named_database(state: &EnvironmentState, name: &str) -> InternalResult<BinaryDatabase> {
if state.options.read_only {
return Err(LifecycleError::ReadOnly);
}
// LMDB requires DBI opens to be serialized within a process. The cache lock
// covers lookup, DBI creation, and commit, then CRUD can use the immutable
// cached handle without taking this lock.
let mut databases = database_cache(state);
if let Some(database) = databases.get(name) {
return Ok(*database);
}
let mut transaction = state
.env
.write_txn()
.map_err(|_| LifecycleError::TransactionFailed)?;
let database = state
.env
.create_database::<Bytes, Bytes>(&mut transaction, Some(name))
.map_err(|_| LifecycleError::DatabaseCreateFailed)?;
transaction
.commit()
.map_err(|_| LifecycleError::TransactionFailed)?;
databases.insert(name.to_owned(), database);
Ok(database)
}
fn open_named_database(state: &EnvironmentState, name: &str) -> InternalResult<BinaryDatabase> {
let mut databases = database_cache(state);
if let Some(database) = databases.get(name) {
return Ok(*database);
}
let transaction = state
.env
.read_txn()
.map_err(|_| LifecycleError::TransactionFailed)?;
let database = state
.env
.open_database::<Bytes, Bytes>(&transaction, Some(name))
.map_err(|_| LifecycleError::DatabaseOpenFailed)?
.ok_or(LifecycleError::DatabaseNotFound)?;
// heed requires committing read transactions which opened a DBI, especially
// when multiple OS processes use the environment.
transaction
.commit()
.map_err(|_| LifecycleError::TransactionFailed)?;
databases.insert(name.to_owned(), database);
Ok(database)
}
fn decode_environment(term: Term<'_>) -> InternalResult<ResourceArc<EnvironmentResource>> {
term.decode()
.map_err(|_| LifecycleError::InvalidEnvironment)
}
fn decode_database(term: Term<'_>) -> InternalResult<ResourceArc<DatabaseResource>> {
term.decode()
.map_err(|_| LifecycleError::InvalidEnvironment)
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum CrudError {
InvalidDatabase,
InvalidKey,
InvalidValue,
ReadOnly,
MapFull,
MapResized,
ReadersFull,
TransactionFull,
OutOfMemory,
Io,
Database,
InvalidBatch,
BatchTooLarge,
MixedEnvironments,
InvalidExpected,
InvalidReplacement,
}
impl CrudError {
fn atom(self) -> rustler::Atom {
match self {
Self::InvalidDatabase => atoms::invalid_database(),
Self::InvalidKey => atoms::invalid_key(),
Self::InvalidValue => atoms::invalid_value(),
Self::ReadOnly => atoms::read_only(),
Self::MapFull => atoms::map_full(),
Self::MapResized => atoms::map_resized(),
Self::ReadersFull => atoms::readers_full(),
Self::TransactionFull => atoms::transaction_full(),
Self::OutOfMemory => atoms::out_of_memory(),
Self::Io => atoms::io_error(),
Self::Database => atoms::database_error(),
Self::InvalidBatch => atoms::invalid_batch(),
Self::BatchTooLarge => atoms::batch_too_large(),
Self::MixedEnvironments => atoms::mixed_environments(),
Self::InvalidExpected => atoms::invalid_expected(),
Self::InvalidReplacement => atoms::invalid_replacement(),
}
}
}
fn map_heed_error(error: heed::Error) -> CrudError {
match error {
heed::Error::Io(error) if error.kind() == std::io::ErrorKind::OutOfMemory => {
CrudError::OutOfMemory
}
heed::Error::Io(_) => CrudError::Io,
heed::Error::Mdb(MdbError::MapFull) => CrudError::MapFull,
heed::Error::Mdb(MdbError::MapResized) => CrudError::MapResized,
heed::Error::Mdb(MdbError::ReadersFull) => CrudError::ReadersFull,
heed::Error::Mdb(MdbError::TxnFull) => CrudError::TransactionFull,
heed::Error::Mdb(MdbError::BadDbi | MdbError::Incompatible) => CrudError::InvalidDatabase,
heed::Error::Mdb(MdbError::BadValSize) => CrudError::InvalidKey,
heed::Error::Mdb(_) | heed::Error::Encoding(_) | heed::Error::Decoding(_) => {
CrudError::Database
}
heed::Error::EnvAlreadyOpened => CrudError::Database,
}
}
fn decode_crud_database(term: Term<'_>) -> Result<ResourceArc<DatabaseResource>, CrudError> {
term.decode().map_err(|_| CrudError::InvalidDatabase)
}
fn decode_binary<'a>(term: Term<'a>, error: CrudError) -> Result<Binary<'a>, CrudError> {
Binary::decode(term).map_err(|_| error)
}
fn validate_key<'a>(state: &EnvironmentState, key: &'a [u8]) -> Result<&'a [u8], CrudError> {
if key.is_empty() || key.len() > state.env.max_key_size() {
Err(CrudError::InvalidKey)
} else {
Ok(key)
}
}
fn put_binary(database: &DatabaseResource, key: &[u8], value: &[u8]) -> Result<(), CrudError> {
validate_key(&database.state, key)?;
if database.state.options.read_only {
return Err(CrudError::ReadOnly);
}
let mut transaction = database.state.env.write_txn().map_err(map_heed_error)?;
database
.database
.put(&mut transaction, key, value)
.map_err(map_heed_error)?;
transaction.commit().map_err(map_heed_error)
}
fn clear_stale_readers(state: &EnvironmentState) -> Result<usize, CrudError> {
state.env.clear_stale_readers().map_err(map_heed_error)
}
fn delete_binary(database: &DatabaseResource, key: &[u8]) -> Result<(), CrudError> {
validate_key(&database.state, key)?;
if database.state.options.read_only {
return Err(CrudError::ReadOnly);
}
let mut transaction = database.state.env.write_txn().map_err(map_heed_error)?;
database
.database
.delete(&mut transaction, key)
.map_err(map_heed_error)?;
transaction.commit().map_err(map_heed_error)
}
fn copy_bytes(bytes: &[u8]) -> Result<Vec<u8>, CrudError> {
let mut copied = Vec::new();
copied
.try_reserve_exact(bytes.len())
.map_err(|_| CrudError::OutOfMemory)?;
copied.extend_from_slice(bytes);
Ok(copied)
}
#[derive(Clone, Debug, Eq, PartialEq)]
enum ScanMode {
Prefix(Vec<u8>),
Range {
from: Option<Vec<u8>>,
to: Option<Vec<u8>>,
from_inclusive: bool,
to_inclusive: bool,
},
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ScanError {
InvalidDatabase,
InvalidMode,
InvalidLimit,
InvalidMaxBytes,
RowTooLarge,
InvalidContinuation,
StaleContinuation,
OutOfMemory,
MapResized,
ReadersFull,
Io,
Database,
}
impl ScanError {
fn atom(self) -> Atom {
match self {
Self::InvalidDatabase => atoms::invalid_database(),
Self::InvalidMode => atoms::invalid_mode(),
Self::InvalidLimit => atoms::invalid_limit(),
Self::InvalidMaxBytes => atoms::invalid_max_bytes(),
Self::RowTooLarge => atoms::row_too_large(),
Self::InvalidContinuation => atoms::invalid_continuation(),
Self::StaleContinuation => atoms::stale_continuation(),
Self::OutOfMemory => atoms::out_of_memory(),
Self::MapResized => atoms::map_resized(),
Self::ReadersFull => atoms::readers_full(),
Self::Io => atoms::io_error(),
Self::Database => atoms::database_error(),
}
}
}
fn map_scan_heed_error(error: heed::Error) -> ScanError {
match map_heed_error(error) {
CrudError::OutOfMemory => ScanError::OutOfMemory,
CrudError::MapResized => ScanError::MapResized,
CrudError::ReadersFull => ScanError::ReadersFull,
CrudError::Io => ScanError::Io,
CrudError::InvalidDatabase => ScanError::InvalidDatabase,
_ => ScanError::Database,
}
}
fn scan_token_tag(bytes: &[u8]) -> u64 {
// RandomState provides a process-local keyed SipHash-family hasher. The
// integrity tag rejects accidental or unsophisticated token modification;
// it is not a cryptographic authentication contract. Tokens remain
// process-local and free of native pointers.
let mut hasher = SCAN_TOKEN_HASHER.build_hasher();
hasher.write(bytes);
hasher.finish()
}
fn push_sized_bytes(output: &mut Vec<u8>, bytes: &[u8]) -> Result<(), ScanError> {
let length = u32::try_from(bytes.len()).map_err(|_| ScanError::InvalidContinuation)?;
output.extend_from_slice(&length.to_be_bytes());
output.extend_from_slice(bytes);
Ok(())
}
fn scan_binding(database_name: &str, mode: &ScanMode) -> Result<Vec<u8>, ScanError> {
let mode_bytes = match mode {
ScanMode::Prefix(prefix) => 1_usize
.checked_add(4)
.and_then(|size| size.checked_add(prefix.len())),
ScanMode::Range { from, to, .. } => {
[from, to].into_iter().try_fold(3_usize, |size, bound| {
let field_size = bound.as_ref().map_or(Some(1), |bytes| {
1_usize.checked_add(4)?.checked_add(bytes.len())
})?;
size.checked_add(field_size)
})
}
}
.ok_or(ScanError::OutOfMemory)?;
let capacity = 4_usize
.checked_add(database_name.len())
.and_then(|size| size.checked_add(mode_bytes))
.ok_or(ScanError::OutOfMemory)?;
let mut binding = Vec::new();
binding
.try_reserve_exact(capacity)
.map_err(|_| ScanError::OutOfMemory)?;
push_sized_bytes(&mut binding, database_name.as_bytes())?;
match mode {
ScanMode::Prefix(prefix) => {
binding.push(0);
push_sized_bytes(&mut binding, prefix)?;
}
ScanMode::Range {
from,
to,
from_inclusive,
to_inclusive,
} => {
binding.push(1);
for bound in [from, to] {
match bound {
None => binding.push(0),
Some(bytes) => {
binding.push(1);
push_sized_bytes(&mut binding, bytes)?;
}
}
}
binding.push(u8::from(*from_inclusive));
binding.push(u8::from(*to_inclusive));
}
}
Ok(binding)
}
fn make_scan_token(
environment_id: u64,
database_name: &str,
mode: &ScanMode,
last_key: &[u8],
) -> Result<Vec<u8>, ScanError> {
let binding = scan_binding(database_name, mode)?;
let capacity = 4_usize
.checked_add(8)
.and_then(|size| size.checked_add(4))
.and_then(|size| size.checked_add(binding.len()))
.and_then(|size| size.checked_add(4))
.and_then(|size| size.checked_add(last_key.len()))
.and_then(|size| size.checked_add(8))
.ok_or(ScanError::OutOfMemory)?;
let mut token = Vec::new();
token
.try_reserve_exact(capacity)
.map_err(|_| ScanError::OutOfMemory)?;
token.extend_from_slice(SCAN_TOKEN_MAGIC);
token.extend_from_slice(&environment_id.to_be_bytes());
push_sized_bytes(&mut token, &binding)?;
push_sized_bytes(&mut token, last_key)?;
let tag = scan_token_tag(&token);
token.extend_from_slice(&tag.to_be_bytes());
Ok(token)
}
fn take_token_field<'a>(token: &'a [u8], offset: &mut usize) -> Result<&'a [u8], ScanError> {
let length_end = offset
.checked_add(4)
.ok_or(ScanError::InvalidContinuation)?;
let length_bytes: [u8; 4] = token
.get(*offset..length_end)
.ok_or(ScanError::InvalidContinuation)?
.try_into()
.map_err(|_| ScanError::InvalidContinuation)?;
*offset = length_end;
let end = offset
.checked_add(u32::from_be_bytes(length_bytes) as usize)
.ok_or(ScanError::InvalidContinuation)?;
let field = token
.get(*offset..end)
.ok_or(ScanError::InvalidContinuation)?;
*offset = end;
Ok(field)
}
fn environment_id_is_live(environment_id: u64) -> bool {
registry().values().any(|state| {
state
.upgrade()
.is_some_and(|state| state.id == environment_id)
})
}
fn scan_mode_contains_key(mode: &ScanMode, key: &[u8]) -> bool {
match mode {
ScanMode::Prefix(prefix) => key.starts_with(prefix),
ScanMode::Range {
from,
to,
from_inclusive,
to_inclusive,
} => {
let above_from = from.as_ref().is_none_or(|from| {
if *from_inclusive {
key >= from.as_slice()
} else {
key > from.as_slice()
}
});
let below_to = to.as_ref().is_none_or(|to| {
if *to_inclusive {
key <= to.as_slice()
} else {
key < to.as_slice()
}
});
above_from && below_to
}
}
}
fn parse_scan_token(
token: &[u8],
environment_id: u64,
database_name: &str,
mode: &ScanMode,
max_key_size: usize,
) -> Result<Vec<u8>, ScanError> {
if token.len() < 4 + 8 + 4 + 4 + 8
|| token.len() > MAX_SCAN_TOKEN_BYTES
|| token.get(..4) != Some(SCAN_TOKEN_MAGIC.as_slice())
{
return Err(ScanError::InvalidContinuation);
}
let payload_end = token.len() - 8;
let expected_tag = u64::from_be_bytes(
token[payload_end..]
.try_into()
.map_err(|_| ScanError::InvalidContinuation)?,
);
if scan_token_tag(&token[..payload_end]) != expected_tag {
return Err(ScanError::InvalidContinuation);
}
let token_environment = u64::from_be_bytes(
token[4..12]
.try_into()
.map_err(|_| ScanError::InvalidContinuation)?,
);
let mut offset = 12;
let binding = take_token_field(&token[..payload_end], &mut offset)?;
let last_key = take_token_field(&token[..payload_end], &mut offset)?;
if offset != payload_end {
return Err(ScanError::InvalidContinuation);
}
if token_environment != environment_id {
return Err(if environment_id_is_live(token_environment) {
ScanError::InvalidContinuation
} else {
ScanError::StaleContinuation
});
}
if binding != scan_binding(database_name, mode)?.as_slice()
|| last_key.is_empty()
|| last_key.len() > max_key_size
|| !scan_mode_contains_key(mode, last_key)
{
return Err(ScanError::InvalidContinuation);
}
copy_bytes(last_key).map_err(|error| match error {
CrudError::OutOfMemory => ScanError::OutOfMemory,
_ => ScanError::Database,
})
}
fn prefix_upper_bound(prefix: &[u8]) -> Result<Option<Vec<u8>>, ScanError> {
let Some(index) = prefix.iter().rposition(|byte| *byte != 0xff) else {
return Ok(None);
};
let upper_len = index.checked_add(1).ok_or(ScanError::OutOfMemory)?;
let mut upper = Vec::new();
upper
.try_reserve_exact(upper_len)
.map_err(|_| ScanError::OutOfMemory)?;
upper.extend_from_slice(&prefix[..=index]);
upper[index] += 1;
Ok(Some(upper))
}
fn scan_bounds_are_empty(start: &Bound<&[u8]>, end: &Bound<&[u8]>) -> bool {
let (start_key, start_inclusive) = match start {
Bound::Unbounded => {
return matches!(end, Bound::Included(key) | Bound::Excluded(key) if key.is_empty());
}
Bound::Included(key) => (*key, true),
Bound::Excluded(key) => (*key, false),
};
let (end_key, end_inclusive) = match end {
Bound::Unbounded => return false,
Bound::Included(key) => (*key, true),
Bound::Excluded(key) => (*key, false),
};
start_key > end_key
|| (start_key == end_key && !(start_inclusive && end_inclusive))
|| end_key.is_empty()
}
#[derive(Debug, Eq, PartialEq)]
struct ScanPage {
rows: Vec<(Vec<u8>, Vec<u8>)>,
continuation: Option<Vec<u8>>,
}
fn scan_binary(
database: &DatabaseResource,
mode: &ScanMode,
limit: u64,
max_bytes: u64,
continuation: Option<&[u8]>,
) -> Result<ScanPage, ScanError> {
if !(1..=MAX_SCAN_LIMIT).contains(&limit) {
return Err(ScanError::InvalidLimit);
}
if !(1..=MAX_SCAN_BYTES).contains(&max_bytes) {
return Err(ScanError::InvalidMaxBytes);
}
let max_key_size = database.state.env.max_key_size();
let resume = continuation
.map(|token| parse_scan_token(token, database.state.id, &database.name, mode, max_key_size))
.transpose()?;
// No valid LMDB key can have an oversized prefix. Parse a supplied token first
// so malformed continuations still have stable invalid-continuation behavior.
if matches!(mode, ScanMode::Prefix(prefix) if prefix.len() > max_key_size) {
return Ok(ScanPage {
rows: Vec::new(),
continuation: None,
});
}
// An oversized lower bound lies immediately after its max-sized prefix in the
// valid-key universe. Its successor is therefore the first possible key; an
// all-FF prefix has no successor and makes the range empty.
let oversized_lower = match mode {
ScanMode::Range {
from: Some(from), ..
} if from.len() > max_key_size => match prefix_upper_bound(&from[..max_key_size])? {
Some(successor) => Some(successor),
None => {
return Ok(ScanPage {
rows: Vec::new(),
continuation: None,
});
}
},
_ => None,
};
// Keep normalized storage alive while heed owns borrowed bounds. Prefix bounds
// are materialized here rather than relying on a prefix API, including all-FF.
let prefix_upper = match mode {
ScanMode::Prefix(prefix) => prefix_upper_bound(prefix)?,
_ => None,
};
let start: Bound<&[u8]> = if let Some(last_key) = resume.as_deref() {
Bound::Excluded(last_key)
} else if let Some(successor) = oversized_lower.as_deref() {
Bound::Included(successor)
} else {
match mode {
ScanMode::Prefix(prefix) if prefix.is_empty() => Bound::Unbounded,
ScanMode::Prefix(prefix) => Bound::Included(prefix.as_slice()),
ScanMode::Range { from: None, .. } => Bound::Unbounded,
ScanMode::Range {
from: Some(from), ..
} if from.is_empty() => Bound::Unbounded,
ScanMode::Range {
from: Some(from),
from_inclusive: true,
..
} => Bound::Included(from.as_slice()),
ScanMode::Range {
from: Some(from), ..
} => Bound::Excluded(from.as_slice()),
}
};
let end: Bound<&[u8]> = match mode {
ScanMode::Prefix(_) => prefix_upper
.as_deref()
.map_or(Bound::Unbounded, Bound::Excluded),
ScanMode::Range { to: None, .. } => Bound::Unbounded,
ScanMode::Range { to: Some(to), .. } if to.len() > max_key_size => {
Bound::Included(&to[..max_key_size])
}
ScanMode::Range {
to: Some(to),
to_inclusive: true,
..
} => Bound::Included(to.as_slice()),
ScanMode::Range { to: Some(to), .. } => Bound::Excluded(to.as_slice()),
};
if scan_bounds_are_empty(&start, &end) {
return Ok(ScanPage {
rows: Vec::new(),
continuation: None,
});
}
let bounds = (start, end);
let transaction = database.state.env.read_txn().map_err(map_scan_heed_error)?;
let mut iterator = database
.database
.range(&transaction, &bounds)
.map_err(map_scan_heed_error)?;
let mut rows = Vec::new();
rows.try_reserve(limit as usize)
.map_err(|_| ScanError::OutOfMemory)?;
let mut used_bytes = 0_u64;
let mut has_more = false;
for entry in iterator.by_ref() {
let (key, value) = entry.map_err(map_scan_heed_error)?;
let row_bytes = u64::try_from(key.len())
.ok()
.and_then(|bytes| bytes.checked_add(value.len() as u64))
.ok_or(ScanError::OutOfMemory)?;
if rows.len() as u64 == limit
|| (!rows.is_empty() && used_bytes.saturating_add(row_bytes) > max_bytes)
{
has_more = true;
break;
}
if row_bytes > max_bytes {
return Err(ScanError::RowTooLarge);
}
rows.push((
copy_bytes(key).map_err(|_| ScanError::OutOfMemory)?,
copy_bytes(value).map_err(|_| ScanError::OutOfMemory)?,
));
used_bytes = used_bytes.saturating_add(row_bytes);
}
drop(iterator);
transaction.commit().map_err(map_scan_heed_error)?;
let continuation = if has_more {
let last_key = rows.last().ok_or(ScanError::RowTooLarge)?.0.as_slice();
Some(make_scan_token(
database.state.id,
&database.name,
mode,
last_key,
)?)
} else {
None
};
Ok(ScanPage { rows, continuation })
}
enum BatchOperation {
Put {
database: BinaryDatabase,
key: Vec<u8>,
value: Vec<u8>,
},
Delete {
database: BinaryDatabase,
key: Vec<u8>,
},
}
fn write_batch_operations(
state: &EnvironmentState,
operations: &[BatchOperation],
) -> Result<(), CrudError> {
if operations.is_empty() {
return Ok(());
}
if state.options.read_only {
return Err(CrudError::ReadOnly);
}
let mut transaction = state.env.write_txn().map_err(map_heed_error)?;
for operation in operations {
match operation {
BatchOperation::Put {
database,
key,
value,
} => database
.put(&mut transaction, key, value)
.map_err(map_heed_error)?,
BatchOperation::Delete { database, key } => {
database
.delete(&mut transaction, key)
.map_err(map_heed_error)?;
}
}
}
transaction.commit().map_err(map_heed_error)
}
fn validate_batch_bounds(operation_count: usize, aggregate_bytes: usize) -> Result<(), CrudError> {
if operation_count > MAX_BATCH_OPERATIONS || aggregate_bytes > MAX_BATCH_BYTES {
Err(CrudError::BatchTooLarge)
} else {
Ok(())
}
}
fn decode_batch_operations(
environment: &EnvironmentResource,
operations: Term<'_>,
) -> Result<Vec<BatchOperation>, CrudError> {
if !operations.is_list() {
return Err(CrudError::InvalidBatch);
}
let mut decoded = Vec::new();
let mut aggregate_bytes = 0_usize;
let mut remainder = operations;
while !remainder.is_empty_list() {
if decoded.len() == MAX_BATCH_OPERATIONS {
return Err(CrudError::BatchTooLarge);
}
let (operation, tail) = remainder
.list_get_cell()
.map_err(|_| CrudError::InvalidBatch)?;
remainder = tail;
decoded.try_reserve(1).map_err(|_| CrudError::OutOfMemory)?;
if let Ok((tag, database, key, value)) = operation.decode::<(Atom, Term, Term, Term)>() {
if tag != atoms::put() {
return Err(CrudError::InvalidBatch);
}
let database = decode_crud_database(database).map_err(|_| CrudError::InvalidBatch)?;
if !Arc::ptr_eq(&database.state, &environment.state) {
return Err(CrudError::MixedEnvironments);
}
let key = decode_binary(key, CrudError::InvalidBatch)?;
let value = decode_binary(value, CrudError::InvalidBatch)?;
validate_key(&environment.state, key.as_slice())?;
aggregate_bytes = aggregate_bytes
.checked_add(key.len())
.and_then(|bytes| bytes.checked_add(value.len()))
.ok_or(CrudError::BatchTooLarge)?;
validate_batch_bounds(decoded.len() + 1, aggregate_bytes)?;
decoded.push(BatchOperation::Put {
database: database.database,
key: copy_bytes(key.as_slice())?,
value: copy_bytes(value.as_slice())?,
});
} else if let Ok((tag, database, key)) = operation.decode::<(Atom, Term, Term)>() {
if tag != atoms::delete() {
return Err(CrudError::InvalidBatch);
}
let database = decode_crud_database(database).map_err(|_| CrudError::InvalidBatch)?;
if !Arc::ptr_eq(&database.state, &environment.state) {
return Err(CrudError::MixedEnvironments);
}
let key = decode_binary(key, CrudError::InvalidBatch)?;
validate_key(&environment.state, key.as_slice())?;
aggregate_bytes = aggregate_bytes
.checked_add(key.len())
.ok_or(CrudError::BatchTooLarge)?;
validate_batch_bounds(decoded.len() + 1, aggregate_bytes)?;
decoded.push(BatchOperation::Delete {
database: database.database,
key: copy_bytes(key.as_slice())?,
});
} else {
return Err(CrudError::InvalidBatch);
}
}
Ok(decoded)
}
enum ExpectedValue<'a> {
Missing,
Value(&'a [u8]),
}
enum ReplacementValue<'a> {
Delete,
Put(&'a [u8]),
}
#[derive(Debug, Eq, PartialEq)]
enum CompareExchangeResult<T> {
Exchanged,
Conflict(Option<T>),
}
fn compare_exchange_binary<T>(
database: &DatabaseResource,
key: &[u8],
expected: &ExpectedValue<'_>,
replacement: &ReplacementValue<'_>,
copy_current: impl FnOnce(&[u8]) -> Result<T, CrudError>,
) -> Result<CompareExchangeResult<T>, CrudError> {
validate_key(&database.state, key)?;
if database.state.options.read_only {
return Err(CrudError::ReadOnly);
}
let mut transaction = database.state.env.write_txn().map_err(map_heed_error)?;
let current = database
.database
.get(&transaction, key)
.map_err(map_heed_error)?;
let matches = match (expected, current) {
(ExpectedValue::Missing, None) => true,
(ExpectedValue::Value(expected), Some(current)) => *expected == current,
_ => false,
};
if !matches {
// LMDB-owned bytes are copied while the transaction is still alive.
return Ok(CompareExchangeResult::Conflict(
current.map(copy_current).transpose()?,
));
}
match replacement {
ReplacementValue::Delete => {
database
.database
.delete(&mut transaction, key)
.map_err(map_heed_error)?;
}
ReplacementValue::Put(value) => database
.database
.put(&mut transaction, key, *value)
.map_err(map_heed_error)?,
}
transaction.commit().map_err(map_heed_error)?;
Ok(CompareExchangeResult::Exchanged)
}
#[cfg(test)]
fn get_binary(database: &DatabaseResource, key: &[u8]) -> Result<Option<Vec<u8>>, CrudError> {
validate_key(&database.state, key)?;
let transaction = database.state.env.read_txn().map_err(map_heed_error)?;
let value = database
.database
.get(&transaction, key)
.map_err(map_heed_error)?
.map(<[u8]>::to_vec);
transaction.commit().map_err(map_heed_error)?;
Ok(value)
}
fn version() -> (&'static str, &'static str) {
("lean_lmdb", env!("CARGO_PKG_VERSION"))
}
#[rustler::nif]
fn native_version() -> (&'static str, &'static str) {
version()
}
#[rustler::nif]
fn system_page_size() -> usize {
page_size::get()
}
#[rustler::nif(schedule = "DirtyIo")]
fn environment_open(
path: String,
map_size: u64,
max_dbs: u32,
max_readers: u32,
read_only: bool,
create: bool,
) -> NifResult<ResourceArc<EnvironmentResource>> {
let options = validate_options(map_size, max_dbs, max_readers, read_only, create)
.map_err(LifecycleError::atom)?;
let path = canonical_environment_path(&path, create).map_err(LifecycleError::atom)?;
let state = open_shared_environment(path, options, create).map_err(LifecycleError::atom)?;
Ok(ResourceArc::new(EnvironmentResource { state }))
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_create<'a>(
environment: Term<'a>,
name: Binary<'a>,
) -> NifResult<ResourceArc<DatabaseResource>> {
let environment = decode_environment(environment).map_err(LifecycleError::atom)?;
let name = validate_database_name(name.as_slice()).map_err(LifecycleError::atom)?;
let database = create_named_database(&environment.state, name).map_err(LifecycleError::atom)?;
Ok(ResourceArc::new(DatabaseResource {
state: Arc::clone(&environment.state),
database,
name: name.to_owned(),
}))
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_open<'a>(
environment: Term<'a>,
name: Binary<'a>,
) -> NifResult<ResourceArc<DatabaseResource>> {
let environment = decode_environment(environment).map_err(LifecycleError::atom)?;
let name = validate_database_name(name.as_slice()).map_err(LifecycleError::atom)?;
let database = open_named_database(&environment.state, name).map_err(LifecycleError::atom)?;
Ok(ResourceArc::new(DatabaseResource {
state: Arc::clone(&environment.state),
database,
name: name.to_owned(),
}))
}
enum LookupResult<'a> {
Found(Binary<'a>),
Missing,
}
impl Encoder for LookupResult<'_> {
fn encode<'a>(&self, env: NifEnv<'a>) -> Term<'a> {
match self {
Self::Found(binary) => binary.encode(env),
Self::Missing => atoms::not_found().encode(env),
}
}
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_get<'a>(database: Term<'a>, key: Term<'a>) -> NifResult<LookupResult<'a>> {
let env = database.get_env();
let database = decode_crud_database(database).map_err(CrudError::atom)?;
let key = decode_binary(key, CrudError::InvalidKey).map_err(CrudError::atom)?;
let key = validate_key(&database.state, key.as_slice()).map_err(CrudError::atom)?;
let transaction = database
.state
.env
.read_txn()
.map_err(map_heed_error)
.map_err(CrudError::atom)?;
let copied = match database
.database
.get(&transaction, key)
.map_err(map_heed_error)
.map_err(CrudError::atom)?
{
Some(value) => {
let mut copied = OwnedBinary::new(value.len()).ok_or_else(atoms::out_of_memory)?;
copied.as_mut_slice().copy_from_slice(value);
Some(copied)
}
None => None,
};
transaction
.commit()
.map_err(map_heed_error)
.map_err(CrudError::atom)?;
Ok(match copied {
Some(binary) => LookupResult::Found(binary.release(env)),
None => LookupResult::Missing,
})
}
fn decode_scan_bound(term: Term<'_>, max_key_size: usize) -> Result<Option<Vec<u8>>, ScanError> {
if matches!(term.decode::<Atom>(), Ok(atom) if atom == atoms::nil()) {
Ok(None)
} else {
let binary = term
.decode::<Binary<'_>>()
.map_err(|_| ScanError::InvalidMode)?;
if binary.len() > max_key_size {
return Err(ScanError::InvalidMode);
}
copy_bytes(binary.as_slice())
.map(Some)
.map_err(|error| match error {
CrudError::OutOfMemory => ScanError::OutOfMemory,
_ => ScanError::InvalidMode,
})
}
}
fn decode_scan_mode(term: Term<'_>, max_key_size: usize) -> Result<ScanMode, ScanError> {
if let Ok((tag, prefix)) = term.decode::<(Atom, Binary<'_>)>() {
if tag != atoms::prefix() || prefix.len() > max_key_size {
return Err(ScanError::InvalidMode);
}
return copy_bytes(prefix.as_slice())
.map(ScanMode::Prefix)
.map_err(|_| ScanError::OutOfMemory);
}
if let Ok((tag, from, to, from_inclusive, to_inclusive)) =
term.decode::<(Atom, Term<'_>, Term<'_>, bool, bool)>()
{
if tag != atoms::range() {
return Err(ScanError::InvalidMode);
}
return Ok(ScanMode::Range {
from: decode_scan_bound(from, max_key_size)?,
to: decode_scan_bound(to, max_key_size)?,
from_inclusive,
to_inclusive,
});
}
Err(ScanError::InvalidMode)
}
enum ScanContinuationNif<'a> {
Done,
Token(Binary<'a>),
}
impl Encoder for ScanContinuationNif<'_> {
fn encode<'a>(&self, env: NifEnv<'a>) -> Term<'a> {
match self {
Self::Done => atoms::done().encode(env),
Self::Token(token) => token.encode(env),
}
}
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_scan<'a>(
database: Term<'a>,
mode: Term<'a>,
limit: Term<'a>,
max_bytes: Term<'a>,
continuation: Term<'a>,
) -> NifResult<(Vec<(Binary<'a>, Binary<'a>)>, ScanContinuationNif<'a>)> {
let env = database.get_env();
let database = decode_crud_database(database).map_err(|_| atoms::invalid_database())?;
let mode =
decode_scan_mode(mode, database.state.env.max_key_size()).map_err(ScanError::atom)?;
let limit = limit.decode::<u64>().map_err(|_| atoms::invalid_limit())?;
if !(1..=MAX_SCAN_LIMIT).contains(&limit) {
return Err(atoms::invalid_limit());
}
let max_bytes = max_bytes
.decode::<u64>()
.map_err(|_| atoms::invalid_max_bytes())?;
if !(1..=MAX_SCAN_BYTES).contains(&max_bytes) {
return Err(atoms::invalid_max_bytes());
}
let continuation_binary = if matches!(continuation.decode::<Atom>(), Ok(atom) if atom == atoms::nil())
{
None
} else {
Some(
continuation
.decode::<Binary<'a>>()
.map_err(|_| atoms::invalid_continuation())?,
)
};
let page = scan_binary(
&database,
&mode,
limit,
max_bytes,
continuation_binary.as_ref().map(Binary::as_slice),
)
.map_err(ScanError::atom)?;
let mut rows = Vec::new();
rows.try_reserve_exact(page.rows.len())
.map_err(|_| atoms::out_of_memory())?;
for (key, value) in page.rows {
let mut key_binary = OwnedBinary::new(key.len()).ok_or_else(atoms::out_of_memory)?;
key_binary.as_mut_slice().copy_from_slice(&key);
let mut value_binary = OwnedBinary::new(value.len()).ok_or_else(atoms::out_of_memory)?;
value_binary.as_mut_slice().copy_from_slice(&value);
rows.push((key_binary.release(env), value_binary.release(env)));
}
let continuation = match page.continuation {
None => ScanContinuationNif::Done,
Some(token) => {
let mut binary = OwnedBinary::new(token.len()).ok_or_else(atoms::out_of_memory)?;
binary.as_mut_slice().copy_from_slice(&token);
ScanContinuationNif::Token(binary.release(env))
}
};
Ok((rows, continuation))
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_put<'a>(database: Term<'a>, key: Term<'a>, value: Term<'a>) -> NifResult<()> {
let database = decode_crud_database(database).map_err(CrudError::atom)?;
let key = decode_binary(key, CrudError::InvalidKey).map_err(CrudError::atom)?;
let value = decode_binary(value, CrudError::InvalidValue).map_err(CrudError::atom)?;
put_binary(&database, key.as_slice(), value.as_slice()).map_err(CrudError::atom)
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_delete<'a>(database: Term<'a>, key: Term<'a>) -> NifResult<()> {
let database = decode_crud_database(database).map_err(CrudError::atom)?;
let key = decode_binary(key, CrudError::InvalidKey).map_err(CrudError::atom)?;
delete_binary(&database, key.as_slice()).map_err(CrudError::atom)
}
#[rustler::nif(schedule = "DirtyIo")]
fn environment_write_batch(environment: Term<'_>, operations: Term<'_>) -> NifResult<()> {
let environment = environment
.decode::<ResourceArc<EnvironmentResource>>()
.map_err(|_| atoms::invalid_environment())?;
let operations = decode_batch_operations(&environment, operations).map_err(CrudError::atom)?;
write_batch_operations(&environment.state, &operations).map_err(CrudError::atom)
}
enum CompareExchangeNifResult<'a> {
Exchanged,
ConflictMissing,
ConflictValue(Binary<'a>),
}
impl Encoder for CompareExchangeNifResult<'_> {
fn encode<'a>(&self, env: NifEnv<'a>) -> Term<'a> {
match self {
Self::Exchanged => atoms::ok().encode(env),
Self::ConflictMissing => (atoms::conflict(), atoms::missing()).encode(env),
Self::ConflictValue(value) => (atoms::conflict(), value).encode(env),
}
}
}
#[rustler::nif(schedule = "DirtyIo")]
fn database_compare_exchange<'a>(
database: Term<'a>,
key: Term<'a>,
expected: Term<'a>,
replacement: Term<'a>,
) -> NifResult<CompareExchangeNifResult<'a>> {
let env = database.get_env();
let database = decode_crud_database(database).map_err(CrudError::atom)?;
let key = decode_binary(key, CrudError::InvalidKey).map_err(CrudError::atom)?;
validate_key(&database.state, key.as_slice()).map_err(CrudError::atom)?;
let expected = if matches!(expected.decode::<Atom>(), Ok(atom) if atom == atoms::missing()) {
ExpectedValue::Missing
} else if let Ok(value) = expected.decode::<Binary<'a>>() {
ExpectedValue::Value(value.as_slice())
} else {
return Err(CrudError::InvalidExpected.atom());
};
let replacement = if matches!(replacement.decode::<Atom>(), Ok(atom) if atom == atoms::delete())
{
ReplacementValue::Delete
} else if let Ok((tag, value)) = replacement.decode::<(Atom, Binary<'a>)>() {
if tag != atoms::put() {
return Err(CrudError::InvalidReplacement.atom());
}
ReplacementValue::Put(value.as_slice())
} else {
return Err(CrudError::InvalidReplacement.atom());
};
match compare_exchange_binary(
&database,
key.as_slice(),
&expected,
&replacement,
|value| {
let mut copied = OwnedBinary::new(value.len()).ok_or(CrudError::OutOfMemory)?;
copied.as_mut_slice().copy_from_slice(value);
Ok(copied)
},
)
.map_err(CrudError::atom)?
{
CompareExchangeResult::Exchanged => Ok(CompareExchangeNifResult::Exchanged),
CompareExchangeResult::Conflict(None) => Ok(CompareExchangeNifResult::ConflictMissing),
CompareExchangeResult::Conflict(Some(value)) => {
Ok(CompareExchangeNifResult::ConflictValue(value.release(env)))
}
}
}
#[rustler::nif(schedule = "DirtyIo")]
fn environment_clear_stale_readers(environment: Term<'_>) -> NifResult<usize> {
let environment = decode_environment(environment).map_err(LifecycleError::atom)?;
clear_stale_readers(&environment.state).map_err(CrudError::atom)
}
#[cfg(all(debug_assertions, feature = "test-support"))]
fn wait_for_test_release(release_path: &str) -> Result<(), rustler::Atom> {
use std::time::{Duration, Instant};
const TEST_TRANSACTION_TIMEOUT: Duration = Duration::from_secs(60);
let deadline = Instant::now()
.checked_add(TEST_TRANSACTION_TIMEOUT)
.ok_or_else(atoms::test_timeout)?;
loop {
if Path::new(release_path).is_file() {
return Ok(());
}
if Instant::now() >= deadline {
return Err(atoms::test_timeout());
}
std::thread::sleep(Duration::from_millis(5));
}
}
#[cfg(all(debug_assertions, feature = "test-support"))]
fn publish_test_marker(marker_path: &str) -> Result<(), rustler::Atom> {
fs::write(marker_path, b"active").map_err(|_| atoms::io_error())
}
/// Test-only coordination NIF. The marker is published only after LMDB has
/// installed the active reader; the transaction is always released by the
/// bounded poll or by process termination.
#[cfg(all(debug_assertions, feature = "test-support"))]
#[rustler::nif(schedule = "DirtyIo")]
fn test_hold_read_transaction<'a>(
database: Term<'a>,
key: Term<'a>,
marker_path: String,
release_path: String,
) -> NifResult<()> {
let database = decode_crud_database(database).map_err(CrudError::atom)?;
let key = decode_binary(key, CrudError::InvalidKey).map_err(CrudError::atom)?;
let key = validate_key(&database.state, key.as_slice()).map_err(CrudError::atom)?;
let transaction = database
.state
.env
.read_txn()
.map_err(map_heed_error)
.map_err(CrudError::atom)?;
let _value = database
.database
.get(&transaction, key)
.map_err(map_heed_error)
.map_err(CrudError::atom)?;
publish_test_marker(&marker_path)?;
wait_for_test_release(&release_path)?;
transaction
.commit()
.map_err(map_heed_error)
.map_err(CrudError::atom)
}
/// Test-only coordination NIF. Acquiring the transaction before publishing the
/// marker proves that the LMDB writer lock is held while the test coordinates.
#[cfg(all(debug_assertions, feature = "test-support"))]
#[rustler::nif(schedule = "DirtyIo")]
fn test_hold_write_transaction<'a>(
database: Term<'a>,
key: Term<'a>,
value: Term<'a>,
marker_path: String,
release_path: String,
) -> NifResult<()> {
let database = decode_crud_database(database).map_err(CrudError::atom)?;
let key = decode_binary(key, CrudError::InvalidKey).map_err(CrudError::atom)?;
let key = validate_key(&database.state, key.as_slice()).map_err(CrudError::atom)?;
let value = decode_binary(value, CrudError::InvalidValue).map_err(CrudError::atom)?;
if database.state.options.read_only {
return Err(atoms::read_only());
}
let mut transaction = database
.state
.env
.write_txn()
.map_err(map_heed_error)
.map_err(CrudError::atom)?;
database
.database
.put(&mut transaction, key, value.as_slice())
.map_err(map_heed_error)
.map_err(CrudError::atom)?;
publish_test_marker(&marker_path)?;
wait_for_test_release(&release_path)?;
transaction
.commit()
.map_err(map_heed_error)
.map_err(CrudError::atom)
}
#[rustler::nif]
fn environment_info(environment: Term<'_>) -> NifResult<(u64, String, usize, u32, u32, bool)> {
let environment = decode_environment(environment).map_err(LifecycleError::atom)?;
let state = &environment.state;
let path = state
.path
.to_str()
.ok_or(LifecycleError::InvalidPath)
.map_err(LifecycleError::atom)?;
Ok((
state.id,
path.to_owned(),
state.options.map_size,
state.options.max_dbs,
state.options.max_readers,
state.options.read_only,
))
}
#[rustler::nif]
fn database_info(database: Term<'_>) -> NifResult<(u64, String)> {
let database = decode_database(database).map_err(LifecycleError::atom)?;
// Touch the immutable DBI handle here so this diagnostic also proves that
// the database resource—not merely its Elixir metadata—decoded correctly.
let _database_handle = database.database;
Ok((database.state.id, database.name.clone()))
}
#[rustler::nif(schedule = "DirtyIo")]
fn registry_info() -> (usize, usize, usize) {
let mut entries = registry();
let total = entries.len();
let live = entries
.values()
.filter(|state| state.strong_count() != 0)
.count();
let stale = total - live;
// Report stale entries once, then ensure the registry cannot accumulate them.
entries.retain(|_, state| state.strong_count() != 0);
(live, stale, total)
}
rustler::init!("Elixir.LeanLmdb.Native");
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use std::sync::{Barrier, Mutex as TestMutex};
use std::thread;
use std::time::{SystemTime, UNIX_EPOCH};
static TEST_LOCK: TestMutex<()> = TestMutex::new(());
fn options() -> EnvironmentOptions {
EnvironmentOptions {
map_size: 10 * 1024 * 1024,
max_dbs: 8,
max_readers: 16,
read_only: false,
}
}
fn fresh_dir(label: &str) -> PathBuf {
let sequence = NEXT_RESOURCE_ID.fetch_add(1, Ordering::Relaxed);
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_nanos());
std::env::temp_dir().join(format!(
"lean_lmdb_{label}_{}_{}_{}",
std::process::id(),
nanos,
sequence
))
}
fn cleanup_environment(path: &Path) {
registry().remove(path);
if let Some(closing) = heed::env_closing_event(path) {
closing.wait();
}
if path.exists() {
fs::remove_dir_all(path).expect("remove isolated test environment");
}
}
fn replace_scan_token_tag(token: &mut Vec<u8>) {
token.truncate(
token
.len()
.checked_sub(8)
.expect("token has an integrity tag"),
);
let tag = scan_token_tag(token);
token.extend_from_slice(&tag.to_be_bytes());
}
#[test]
fn reports_the_crate_version() {
assert_eq!(version(), ("lean_lmdb", "0.1.0"));
}
#[test]
fn validates_environment_option_bounds() {
for (map_size, max_dbs, max_readers) in [(MIN_MAP_SIZE, 1, 1), (MAX_MAP_SIZE, 1024, 4096)] {
assert!(validate_options(map_size, max_dbs, max_readers, false, true).is_ok());
}
for (map_size, max_dbs, max_readers) in [
(MIN_MAP_SIZE - 4096, 8, 16),
(MIN_MAP_SIZE + 1, 8, 16),
(MAX_MAP_SIZE + 4096, 8, 16),
(MIN_MAP_SIZE, 0, 16),
(MIN_MAP_SIZE, 1025, 16),
(MIN_MAP_SIZE, 8, 0),
(MIN_MAP_SIZE, 8, 4097),
] {
assert_eq!(
validate_options(map_size, max_dbs, max_readers, false, false),
Err(LifecycleError::InvalidEnvironmentOptions)
);
}
assert_eq!(
validate_options(MIN_MAP_SIZE, 8, 16, true, true),
Err(LifecycleError::InvalidEnvironmentOptions)
);
}
#[test]
fn validates_database_names() {
assert_eq!(validate_database_name(b"named-db"), Ok("named-db"));
assert!(validate_database_name(b"").is_err());
assert!(validate_database_name(&vec![b'a'; 512]).is_err());
assert!(validate_database_name(b"has\0nul").is_err());
assert!(validate_database_name(&[0xff]).is_err());
assert!(validate_database_name(&vec![b'a'; 511]).is_ok());
}
#[test]
fn canonical_paths_collapse_equivalent_spellings() {
let dir = fresh_dir("canonical");
fs::create_dir_all(&dir).expect("create isolated test directory");
let direct = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), false)
.expect("canonical path");
let equivalent =
canonical_environment_path(dir.join(".").to_str().expect("UTF-8 temp path"), false)
.expect("equivalent canonical path");
assert_eq!(direct, equivalent);
fs::remove_dir_all(dir).expect("remove isolated test directory");
}
#[test]
fn shared_registry_reuses_compatible_state_and_rejects_incompatible_options() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("reuse");
let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
.expect("canonical path");
let first =
open_shared_environment(canonical.clone(), options(), true).expect("first open");
let second =
open_shared_environment(canonical.clone(), options(), true).expect("shared open");
assert!(Arc::ptr_eq(&first, &second));
assert_eq!(first.id, second.id);
let incompatible = EnvironmentOptions {
max_readers: options().max_readers + 1,
..options()
};
assert_eq!(
open_shared_environment(canonical.clone(), incompatible, true).err(),
Some(LifecycleError::IncompatibleEnvironmentOptions)
);
drop(second);
drop(first);
cleanup_environment(&canonical);
}
#[test]
fn concurrent_opens_create_exactly_one_shared_state() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("concurrent");
let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
.expect("canonical path");
let barrier = Arc::new(Barrier::new(8));
let mut workers = Vec::new();
for _ in 0..8 {
let barrier = Arc::clone(&barrier);
let path = canonical.clone();
workers.push(thread::spawn(move || {
barrier.wait();
open_shared_environment(path, options(), true).expect("concurrent open")
}));
}
let states: Vec<_> = workers
.into_iter()
.map(|worker| worker.join().expect("worker did not panic"))
.collect();
assert!(states.iter().all(|state| state.id == states[0].id));
drop(states);
cleanup_environment(&canonical);
}
#[test]
fn named_databases_are_created_reopened_and_missing_is_stable() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("databases");
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");
create_named_database(&state, "first").expect("create first database");
create_named_database(&state, "second").expect("create second database");
open_named_database(&state, "first").expect("reopen first database");
open_named_database(&state, "second").expect("reopen second database");
assert_eq!(
open_named_database(&state, "missing").err(),
Some(LifecycleError::DatabaseNotFound)
);
drop(state);
cleanup_environment(&canonical);
}
fn database_resource(state: &Arc<EnvironmentState>, name: &str) -> DatabaseResource {
DatabaseResource {
state: Arc::clone(state),
database: create_named_database(state, name).expect("create database"),
name: name.to_owned(),
}
}
#[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 scan_tokens_are_versioned_bound_and_authenticated() {
let mode = ScanMode::Range {
from: Some(vec![0, 1]),
to: None,
from_inclusive: false,
to_inclusive: true,
};
let token = make_scan_token(42, "data", &mode, &[0xff, 0]).expect("make token");
assert_eq!(
parse_scan_token(&token, 42, "data", &mode, 511),
Ok(vec![0xff, 0])
);
let stale_token =
make_scan_token(u64::MAX, "data", &mode, &[0xff, 0]).expect("make stale token");
assert_eq!(
parse_scan_token(&stale_token, 42, "data", &mode, 511),
Err(ScanError::StaleContinuation)
);
assert_eq!(
parse_scan_token(&token, 42, "other", &mode, 511),
Err(ScanError::InvalidContinuation)
);
assert_eq!(
parse_scan_token(&token, 42, "data", &ScanMode::Prefix(vec![0, 1]), 511,),
Err(ScanError::InvalidContinuation)
);
let empty_key_token =
make_scan_token(42, "data", &mode, b"").expect("make malformed token");
assert_eq!(
parse_scan_token(&empty_key_token, 42, "data", &mode, 511),
Err(ScanError::InvalidContinuation)
);
let prefix_mode = ScanMode::Prefix(b"ab".to_vec());
let outside_prefix =
make_scan_token(42, "data", &prefix_mode, b"ac").expect("make out-of-prefix token");
assert_eq!(
parse_scan_token(&outside_prefix, 42, "data", &prefix_mode, 511),
Err(ScanError::InvalidContinuation)
);
let bounded_mode = ScanMode::Range {
from: Some(b"b".to_vec()),
to: Some(b"d".to_vec()),
from_inclusive: false,
to_inclusive: false,
};
for outside_key in [&b"b"[..], &b"d"[..], &b"z"[..]] {
let outside_range = make_scan_token(42, "data", &bounded_mode, outside_key)
.expect("make out-of-range token");
assert_eq!(
parse_scan_token(&outside_range, 42, "data", &bounded_mode, 511),
Err(ScanError::InvalidContinuation)
);
}
let mut corrupt = token.clone();
corrupt[15] ^= 0x80;
assert_eq!(
parse_scan_token(&corrupt, 42, "data", &mode, 511),
Err(ScanError::InvalidContinuation)
);
assert_eq!(
parse_scan_token(&token[..token.len() - 1], 42, "data", &mode, 511),
Err(ScanError::InvalidContinuation)
);
let mut trailing = token.clone();
trailing.insert(trailing.len() - 8, 0);
replace_scan_token_tag(&mut trailing);
assert_eq!(
parse_scan_token(&trailing, 42, "data", &mode, 511),
Err(ScanError::InvalidContinuation)
);
let mut oversized_binding_length = token.clone();
oversized_binding_length[12..16].copy_from_slice(&u32::MAX.to_be_bytes());
replace_scan_token_tag(&mut oversized_binding_length);
assert_eq!(
parse_scan_token(&oversized_binding_length, 42, "data", &mode, 511),
Err(ScanError::InvalidContinuation)
);
let binding_length = u32::from_be_bytes(token[12..16].try_into().expect("binding length"));
let last_key_length_offset = 16 + binding_length as usize;
let mut oversized_key_length = token.clone();
oversized_key_length[last_key_length_offset..last_key_length_offset + 4]
.copy_from_slice(&u32::MAX.to_be_bytes());
replace_scan_token_tag(&mut oversized_key_length);
assert_eq!(
parse_scan_token(&oversized_key_length, 42, "data", &mode, 511),
Err(ScanError::InvalidContinuation)
);
}
#[test]
fn continuation_environment_mismatch_distinguishes_live_from_stale() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let first_dir = fresh_dir("token_first_env");
let first_path =
canonical_environment_path(first_dir.to_str().expect("UTF-8 temp path"), true)
.expect("first canonical path");
let first = open_shared_environment(first_path.clone(), options(), true)
.expect("first environment");
let second_dir = fresh_dir("token_second_env");
let second_path =
canonical_environment_path(second_dir.to_str().expect("UTF-8 temp path"), true)
.expect("second canonical path");
let second = open_shared_environment(second_path.clone(), options(), true)
.expect("second environment");
let mode = ScanMode::Prefix(Vec::new());
let token = make_scan_token(first.id, "data", &mode, b"key").expect("make token");
assert_eq!(
parse_scan_token(&token, second.id, "data", &mode, second.env.max_key_size()),
Err(ScanError::InvalidContinuation)
);
drop(first);
cleanup_environment(&first_path);
assert_eq!(
parse_scan_token(&token, second.id, "data", &mode, second.env.max_key_size()),
Err(ScanError::StaleContinuation)
);
drop(second);
cleanup_environment(&second_path);
}
#[test]
fn prefix_successors_cover_empty_and_all_ff_prefixes() {
assert_eq!(prefix_upper_bound(b""), Ok(None));
assert_eq!(prefix_upper_bound(&[0xff]), Ok(None));
assert_eq!(prefix_upper_bound(&[0xff, 0xff]), Ok(None));
assert_eq!(prefix_upper_bound(b"abc"), Ok(Some(b"abd".to_vec())));
assert_eq!(
prefix_upper_bound(&[0x12, 0xff, 0xff]),
Ok(Some(vec![0x13]))
);
assert_eq!(
prefix_upper_bound(&[0x12, 0xfe, 0xff]),
Ok(Some(vec![0x12, 0xff]))
);
}
#[test]
fn scans_prefixes_in_pages_and_resumes_exclusively() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("prefix_scan");
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 = database_resource(&state, "data");
for (key, value) in [
(&b"a"[..], &b"zero"[..]),
(&b"aa"[..], &b"one"[..]),
(&b"ab"[..], &b"two"[..]),
(&b"b"[..], &b"three"[..]),
(&[0xff][..], &b"ff"[..]),
(&[0xff, 0][..], &b"ff-zero"[..]),
] {
put_binary(&database, key, value).expect("seed scan row");
}
let mode = ScanMode::Prefix(b"a".to_vec());
let first = scan_binary(&database, &mode, 2, MAX_SCAN_BYTES, None).expect("first page");
assert_eq!(
first.rows,
vec![
(b"a".to_vec(), b"zero".to_vec()),
(b"aa".to_vec(), b"one".to_vec())
]
);
let token = first.continuation.expect("more prefix rows");
// Limits are intentionally excluded from token binding.
let second = scan_binary(&database, &mode, 10, 5, Some(&token)).expect("second page");
assert_eq!(second.rows, vec![(b"ab".to_vec(), b"two".to_vec())]);
assert_eq!(second.continuation, None);
let all_ff = scan_binary(
&database,
&ScanMode::Prefix(vec![0xff]),
10,
MAX_SCAN_BYTES,
None,
)
.expect("all-FF prefix");
assert_eq!(
all_ff.rows,
vec![
(vec![0xff], b"ff".to_vec()),
(vec![0xff, 0], b"ff-zero".to_vec())
]
);
let all = scan_binary(
&database,
&ScanMode::Prefix(Vec::new()),
10,
MAX_SCAN_BYTES,
None,
)
.expect("empty prefix");
assert_eq!(all.rows.len(), 6);
drop(database);
drop(state);
cleanup_environment(&canonical);
}
#[test]
fn range_scan_honors_bounds_lookahead_byte_budget_and_limits() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("range_scan");
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 = database_resource(&state, "data");
for key in [b"a", b"b", b"c", b"d"] {
let value = if key == b"b" {
vec![7; 20]
} else {
key.to_vec()
};
put_binary(&database, key, &value).expect("seed range row");
}
let mode = ScanMode::Range {
from: Some(b"a".to_vec()),
to: Some(b"d".to_vec()),
from_inclusive: false,
to_inclusive: false,
};
assert_eq!(
scan_binary(&database, &mode, 10, 2, None),
Err(ScanError::RowTooLarge)
);
let first = scan_binary(&database, &mode, 10, 21, None).expect("bounded first row");
assert_eq!(first.rows, vec![(b"b".to_vec(), vec![7; 20])]);
let token = first.continuation.expect("lookahead found c");
let second = scan_binary(&database, &mode, 10, 2, Some(&token)).expect("resume range");
assert_eq!(second.rows, vec![(b"c".to_vec(), b"c".to_vec())]);
assert_eq!(second.continuation, None);
assert_eq!(
scan_binary(&database, &mode, 0, 1, None),
Err(ScanError::InvalidLimit)
);
assert_eq!(
scan_binary(&database, &mode, MAX_SCAN_LIMIT + 1, 1, None),
Err(ScanError::InvalidLimit)
);
assert_eq!(
scan_binary(&database, &mode, 1, 0, None),
Err(ScanError::InvalidMaxBytes)
);
assert_eq!(
scan_binary(&database, &mode, 1, MAX_SCAN_BYTES + 1, None),
Err(ScanError::InvalidMaxBytes)
);
drop(database);
drop(state);
cleanup_environment(&canonical);
}
#[test]
fn scans_normalize_oversized_prefixes_and_range_bounds() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("oversized_scan_bounds");
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 = database_resource(&state, "data");
let max_key_size = state.env.max_key_size();
let max_prefix = vec![0x10; max_key_size];
let mut successor = max_prefix.clone();
*successor.last_mut().expect("nonzero max key size") += 1;
let high = vec![0x20; max_key_size];
for key in [
vec![0x01],
max_prefix.clone(),
successor.clone(),
high.clone(),
] {
put_binary(&database, &key, &key).expect("seed boundary key");
}
let mut oversized = max_prefix.clone();
oversized.push(0);
for from_inclusive in [false, true] {
let mode = ScanMode::Range {
from: Some(oversized.clone()),
to: None,
from_inclusive,
to_inclusive: false,
};
assert_eq!(
collect_scan_pages(&database, &mode, 10, MAX_SCAN_BYTES),
vec![
(successor.clone(), successor.clone()),
(high.clone(), high.clone())
]
);
}
for to_inclusive in [false, true] {
let mode = ScanMode::Range {
from: None,
to: Some(oversized.clone()),
from_inclusive: false,
to_inclusive,
};
assert_eq!(
collect_scan_pages(&database, &mode, 10, MAX_SCAN_BYTES),
vec![
(vec![0x01], vec![0x01]),
(max_prefix.clone(), max_prefix.clone())
]
);
}
let oversized_prefix = ScanMode::Prefix(vec![0x10; max_key_size + 1]);
assert_eq!(
scan_binary(&database, &oversized_prefix, 10, MAX_SCAN_BYTES, None),
Ok(ScanPage {
rows: Vec::new(),
continuation: None
})
);
let mut ff_tail = vec![0x10; max_key_size];
*ff_tail.last_mut().expect("nonzero max key size") = 0xff;
let ff_successor = prefix_upper_bound(&ff_tail)
.expect("FF-tail successor")
.expect("FF-tail is not all FF");
put_binary(&database, &ff_tail, b"ff-tail").expect("seed FF-tail key");
put_binary(&database, &ff_successor, b"ff-successor").expect("seed FF successor");
let mut ff_tail_oversized = ff_tail.clone();
ff_tail_oversized.push(0);
let ff_lower = ScanMode::Range {
from: Some(ff_tail_oversized.clone()),
to: None,
from_inclusive: true,
to_inclusive: false,
};
let ff_lower_rows = collect_scan_pages(&database, &ff_lower, 10, MAX_SCAN_BYTES);
assert_eq!(ff_lower_rows.first().map(|row| &row.0), Some(&ff_successor));
let ff_upper = ScanMode::Range {
from: None,
to: Some(ff_tail_oversized),
from_inclusive: true,
to_inclusive: false,
};
let ff_upper_rows = collect_scan_pages(&database, &ff_upper, 10, MAX_SCAN_BYTES);
assert!(ff_upper_rows.iter().any(|row| row.0 == ff_tail));
assert!(!ff_upper_rows.iter().any(|row| row.0 == ff_successor));
let mut beyond_all_keys = vec![0xff; max_key_size];
beyond_all_keys.push(0);
let empty_range = ScanMode::Range {
from: Some(beyond_all_keys.clone()),
to: None,
from_inclusive: true,
to_inclusive: false,
};
assert_eq!(
scan_binary(&database, &empty_range, 10, MAX_SCAN_BYTES, None),
Ok(ScanPage {
rows: Vec::new(),
continuation: None
})
);
let all_keys_upper = ScanMode::Range {
from: None,
to: Some(beyond_all_keys),
from_inclusive: true,
to_inclusive: false,
};
assert_eq!(
collect_scan_pages(&database, &all_keys_upper, 10, MAX_SCAN_BYTES),
collect_scan_pages(
&database,
&ScanMode::Range {
from: None,
to: None,
from_inclusive: true,
to_inclusive: false,
},
10,
MAX_SCAN_BYTES
)
);
drop(database);
drop(state);
cleanup_environment(&canonical);
}
fn model_scan(
entries: &BTreeMap<Vec<u8>, Vec<u8>>,
mode: &ScanMode,
) -> Vec<(Vec<u8>, Vec<u8>)> {
entries
.iter()
.filter(|(key, _)| match mode {
ScanMode::Prefix(prefix) => key.starts_with(prefix),
ScanMode::Range {
from,
to,
from_inclusive,
to_inclusive,
} => {
let above_from = from.as_ref().is_none_or(|from| {
if *from_inclusive {
key.as_slice() >= from.as_slice()
} else {
key.as_slice() > from.as_slice()
}
});
let below_to = to.as_ref().is_none_or(|to| {
if *to_inclusive {
key.as_slice() <= to.as_slice()
} else {
key.as_slice() < to.as_slice()
}
});
above_from && below_to
}
})
.map(|(key, value)| (key.clone(), value.clone()))
.collect()
}
fn collect_scan_pages(
database: &DatabaseResource,
mode: &ScanMode,
limit: u64,
max_bytes: u64,
) -> Vec<(Vec<u8>, Vec<u8>)> {
let mut rows = Vec::new();
let mut continuation = None;
for _page_index in 0..20_000 {
let resumed = continuation.is_some();
let page = scan_binary(database, mode, limit, max_bytes, continuation.as_deref())
.expect("model scan page");
if resumed {
assert!(!page.rows.is_empty(), "a continuation must make progress");
}
rows.extend(page.rows);
match page.continuation {
Some(token) => continuation = Some(token),
None => return rows,
}
}
panic!("model scan did not finish within the deterministic page bound")
}
#[test]
fn range_scan_matches_binary_order_model_for_every_boundary_combination() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("range_model");
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 = database_resource(&state, "data");
// Fixed edge cases plus deterministic generated bytes exercise LMDB's unsigned,
// byte-exact lexicographic ordering (including NUL, high-bit, and FF bytes).
let mut model = BTreeMap::new();
let edge_keys = [
vec![0x00],
vec![0x00, 0x00],
vec![0x00, 0xff],
vec![0x01],
vec![0x7f, 0xff],
vec![0x80],
vec![0x80, 0x00],
vec![0xfe, 0xff],
vec![0xff],
vec![0xff, 0x00],
vec![0xff, 0xff],
];
for (index, key) in edge_keys.into_iter().enumerate() {
model.insert(key, vec![0x00, index as u8, 0xff]);
}
let mut seed = 0x6d_2b_79_f5_u32;
for index in 0..48_u8 {
seed = seed.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
let length = (seed as usize % 6) + 1;
let mut key = Vec::with_capacity(length);
for _ in 0..length {
seed = seed.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
key.push((seed >> 24) as u8);
}
model.insert(key, vec![index, 0x00, 0x80, 0xff]);
}
for (key, value) in &model {
put_binary(&database, key, value).expect("seed model row");
}
// None and exact byte edges cover unbounded, equal, empty, and inverted
// ranges. Both booleans are varied even where an unbounded side ignores it.
let bounds = [
None,
Some(Vec::new()),
Some(vec![0x00]),
Some(vec![0x00, 0xff]),
Some(vec![0x80]),
Some(vec![0xff]),
Some(vec![0xff, 0xff]),
Some(vec![0xff, 0xff, 0xff]),
];
for from in &bounds {
for to in &bounds {
for from_inclusive in [false, true] {
for to_inclusive in [false, true] {
let mode = ScanMode::Range {
from: from.clone(),
to: to.clone(),
from_inclusive,
to_inclusive,
};
let expected = model_scan(&model, &mode);
let actual = collect_scan_pages(&database, &mode, 1, 16);
assert_eq!(
actual, expected,
"range mismatch: {mode:?}; each bounded page must return every row exactly once"
);
}
}
}
}
drop(database);
drop(state);
cleanup_environment(&canonical);
}
#[test]
fn continuation_has_documented_weak_snapshot_behavior_across_writes() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("scan_weak_snapshot");
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 = database_resource(&state, "data");
for key in [vec![0x10], vec![0x20], vec![0x30], vec![0x40]] {
put_binary(&database, &key, &key).expect("seed weak snapshot row");
}
let mode = ScanMode::Range {
from: Some(vec![0x10]),
to: Some(vec![0x50]),
from_inclusive: true,
to_inclusive: false,
};
let first = scan_binary(&database, &mode, 1, MAX_SCAN_BYTES, None).expect("first page");
assert_eq!(first.rows, vec![(vec![0x10], vec![0x10])]);
let token = first.continuation.expect("remaining rows");
// A continuation is an exclusive last-key cursor, not a pinned snapshot:
// writes behind it are not revisited, while subsequent pages see later writes.
put_binary(&database, &[0x05], b"behind-range").expect("insert below range");
put_binary(&database, &[0x08], b"behind-cursor").expect("insert behind cursor");
put_binary(&database, &[0x18], b"new").expect("insert ahead of cursor");
put_binary(&database, &[0x30], b"updated").expect("update ahead of cursor");
delete_binary(&database, &[0x20]).expect("delete ahead of cursor");
put_binary(&database, &[0x50], b"outside").expect("insert at excluded upper edge");
let mut actual = first.rows;
let mut continuation = Some(token);
while let Some(token) = continuation {
let page = scan_binary(&database, &mode, 1, 16, Some(&token)).expect("changed page");
actual.extend(page.rows);
continuation = page.continuation;
}
assert_eq!(
actual,
vec![
(vec![0x10], vec![0x10]),
(vec![0x18], b"new".to_vec()),
(vec![0x30], b"updated".to_vec()),
(vec![0x40], vec![0x40]),
]
);
drop(database);
drop(state);
cleanup_environment(&canonical);
}
#[test]
fn batch_bounds_are_checked_before_execution() {
assert_eq!(validate_batch_bounds(0, 0), Ok(()));
assert_eq!(
validate_batch_bounds(MAX_BATCH_OPERATIONS, MAX_BATCH_BYTES),
Ok(())
);
assert_eq!(
validate_batch_bounds(MAX_BATCH_OPERATIONS + 1, 0),
Err(CrudError::BatchTooLarge)
);
assert_eq!(
validate_batch_bounds(1, MAX_BATCH_BYTES + 1),
Err(CrudError::BatchTooLarge)
);
}
#[test]
fn ordered_batch_spans_databases_and_missing_deletes_are_idempotent() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("ordered_batch");
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");
put_binary(&first, b"old", b"present").expect("initial put");
let operations = vec![
BatchOperation::Put {
database: first.database,
key: b"same".to_vec(),
value: b"one".to_vec(),
},
BatchOperation::Put {
database: second.database,
key: vec![0xff],
value: vec![0, 0xff],
},
BatchOperation::Put {
database: first.database,
key: b"same".to_vec(),
value: b"two".to_vec(),
},
BatchOperation::Delete {
database: first.database,
key: b"same".to_vec(),
},
BatchOperation::Delete {
database: first.database,
key: b"missing".to_vec(),
},
BatchOperation::Delete {
database: first.database,
key: b"old".to_vec(),
},
];
write_batch_operations(&state, &operations).expect("atomic batch");
assert_eq!(get_binary(&first, b"same"), Ok(None));
assert_eq!(get_binary(&first, b"old"), Ok(None));
assert_eq!(get_binary(&second, &[0xff]), Ok(Some(vec![0, 0xff])));
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_batch_rolls_back_operations_across_databases() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("map_full_batch");
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 first = database_resource(&state, "first");
let second = database_resource(&state, "second");
put_binary(&first, b"one", b"before-one").expect("initial first value");
put_binary(&second, b"two", b"before-two").expect("initial second value");
let operations = vec![
BatchOperation::Put {
database: first.database,
key: b"one".to_vec(),
value: b"changed".to_vec(),
},
BatchOperation::Delete {
database: second.database,
key: b"two".to_vec(),
},
BatchOperation::Put {
database: first.database,
key: b"large".to_vec(),
value: vec![7; 2 * MIN_MAP_SIZE as usize],
},
];
assert_eq!(
write_batch_operations(&state, &operations),
Err(CrudError::MapFull)
);
assert_eq!(get_binary(&first, b"one"), Ok(Some(b"before-one".to_vec())));
assert_eq!(
get_binary(&second, b"two"),
Ok(Some(b"before-two".to_vec()))
);
assert_eq!(get_binary(&first, b"large"), Ok(None));
drop(first);
drop(second);
drop(state);
cleanup_environment(&canonical);
}
#[test]
fn compare_exchange_handles_empty_missing_delete_and_concurrent_contenders() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("compare_exchange");
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"));
assert_eq!(
compare_exchange_binary(
&database,
b"empty",
&ExpectedValue::Missing,
&ReplacementValue::Put(&[]),
copy_bytes,
),
Ok(CompareExchangeResult::Exchanged)
);
assert_eq!(
compare_exchange_binary(
&database,
b"empty",
&ExpectedValue::Missing,
&ReplacementValue::Delete,
copy_bytes,
),
Ok(CompareExchangeResult::Conflict(Some(Vec::new())))
);
assert_eq!(
compare_exchange_binary(
&database,
b"empty",
&ExpectedValue::Value(&[]),
&ReplacementValue::Delete,
copy_bytes,
),
Ok(CompareExchangeResult::Exchanged)
);
assert_eq!(get_binary(&database, b"empty"), Ok(None));
let barrier = Arc::new(Barrier::new(12));
let mut workers = Vec::new();
for index in 0..12_u8 {
let database = Arc::clone(&database);
let barrier = Arc::clone(&barrier);
workers.push(thread::spawn(move || {
barrier.wait();
let value = vec![0xff, index];
let result = compare_exchange_binary(
&database,
b"contended",
&ExpectedValue::Missing,
&ReplacementValue::Put(value.as_slice()),
copy_bytes,
);
(value, result)
}));
}
let results: Vec<_> = workers
.into_iter()
.map(|worker| worker.join().expect("CAS worker did not panic"))
.collect();
let winners: Vec<_> = results
.iter()
.filter_map(|(value, result)| {
(result == &Ok(CompareExchangeResult::Exchanged)).then_some(value.clone())
})
.collect();
assert_eq!(winners.len(), 1);
let winner = &winners[0];
assert!(results.iter().all(|(value, result)| {
(value == winner && result == &Ok(CompareExchangeResult::Exchanged))
|| result == &Ok(CompareExchangeResult::Conflict(Some(winner.clone())))
}));
assert_eq!(
get_binary(&database, b"contended"),
Ok(Some(winner.clone()))
);
drop(database);
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);
}
#[test]
fn read_only_create_false_requires_existing_data_file() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("read_only_missing");
let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
.expect("canonical path");
let read_only = EnvironmentOptions {
read_only: true,
..options()
};
assert_eq!(
open_shared_environment(canonical.clone(), read_only, false).err(),
Some(LifecycleError::PathNotFound)
);
cleanup_environment(&canonical);
}
#[test]
fn stale_registry_entries_are_replaced_with_new_monotonic_ids() {
let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
let dir = fresh_dir("stale");
let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
.expect("canonical path");
let first_id = open_shared_environment(canonical.clone(), options(), true)
.expect("first open")
.id;
assert_eq!(registry().get(&canonical).map(Weak::strong_count), Some(0));
let second_id = open_shared_environment(canonical.clone(), options(), true)
.expect("reopen")
.id;
assert!(second_id > first_id);
cleanup_environment(&canonical);
}
}