surrealmx 0.20.0

// Copyright © SurrealDB Ltd
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! This module stores the core in-memory database type.

use crate::inner::Inner;
use crate::options::DatabaseOptions;
#[cfg(not(target_arch = "wasm32"))]
use crate::options::{
	DEFAULT_CLEANUP_INTERVAL, DEFAULT_GC_FULL_SCAN_FREQUENCY, DEFAULT_GC_INTERVAL,
};
#[cfg(not(target_arch = "wasm32"))]
use crate::persistence::Persistence;
use crate::pool::Pool;
use crate::pool::DEFAULT_POOL_SIZE;
use crate::tx::Transaction;
use std::ops::Deref;
use std::sync::atomic::{fence, Ordering};
use std::sync::Arc;
use std::time::Duration;

// --------------------------------------------------
// Database
// --------------------------------------------------

/// A transactional in-memory database
pub struct Database {
	/// The inner structure of the database
	inner: Arc<Inner>,
	/// The database transaction pool
	pool: Arc<Pool>,
	/// Optional persistence configuration
	#[cfg(not(target_arch = "wasm32"))]
	persistence: Option<Persistence>,
	/// Interval used by the garbage collector thread
	#[cfg(not(target_arch = "wasm32"))]
	gc_interval: Duration,
	/// Number of gc wake-ups between full datastore scans
	#[cfg(not(target_arch = "wasm32"))]
	gc_full_scan_frequency: u64,
	/// Interval used by the cleanup thread
	#[cfg(not(target_arch = "wasm32"))]
	cleanup_interval: Duration,
}

impl Default for Database {
	fn default() -> Self {
		let inner = Arc::new(Inner::default());
		let pool = Pool::new(inner.clone(), DEFAULT_POOL_SIZE);
		Database {
			inner,
			pool,
			#[cfg(not(target_arch = "wasm32"))]
			persistence: None,
			#[cfg(not(target_arch = "wasm32"))]
			gc_interval: DEFAULT_GC_INTERVAL,
			#[cfg(not(target_arch = "wasm32"))]
			gc_full_scan_frequency: DEFAULT_GC_FULL_SCAN_FREQUENCY,
			#[cfg(not(target_arch = "wasm32"))]
			cleanup_interval: DEFAULT_CLEANUP_INTERVAL,
		}
	}
}

impl Drop for Database {
	fn drop(&mut self) {
		self.shutdown();
	}
}

impl Deref for Database {
	type Target = Inner;

	fn deref(&self) -> &Self::Target {
		&self.inner
	}
}

impl Database {
	/// Create a new transactional in-memory database
	pub fn new() -> Self {
		Self::new_with_options(DatabaseOptions::default())
	}

	/// Create a new transactional in-memory database with custom options
	pub fn new_with_options(opts: DatabaseOptions) -> Self {
		//  Create a new inner database
		let inner = Arc::new(Inner::new(&opts));
		// Initialise a transaction pool
		let pool = Pool::new(inner.clone(), opts.pool_size);
		// Create the database
		let db = Database {
			inner,
			pool,
			#[cfg(not(target_arch = "wasm32"))]
			persistence: None,
			#[cfg(not(target_arch = "wasm32"))]
			gc_interval: opts.gc_interval,
			#[cfg(not(target_arch = "wasm32"))]
			gc_full_scan_frequency: opts.gc_full_scan_frequency,
			#[cfg(not(target_arch = "wasm32"))]
			cleanup_interval: opts.cleanup_interval,
		};
		// Start background tasks when enabled
		#[cfg(not(target_arch = "wasm32"))]
		{
			if opts.enable_cleanup {
				db.initialise_cleanup_worker();
			}
			if opts.enable_gc {
				db.initialise_garbage_worker();
			}
		}
		// Return the database
		db
	}

	/// Create a new persistent database with custom options and persistence
	/// settings
	#[cfg(not(target_arch = "wasm32"))]
	pub fn new_with_persistence(
		opts: DatabaseOptions,
		persistence_opts: crate::PersistenceOptions,
	) -> std::io::Result<Self> {
		//  Create a new inner database
		let inner = Arc::new(Inner::new(&opts));
		// Initialise a transaction pool
		let pool = Pool::new(inner.clone(), opts.pool_size);
		// Create a new persistence layer with options
		let persist = Persistence::new_with_options(persistence_opts, inner.clone())
			.map_err(std::io::Error::other)?;
		// Replace the persistence layer in the database
		inner.persistence.write().replace(Arc::new(persist.clone()));
		// Create the database
		let db = Database {
			inner,
			pool,
			persistence: Some(persist),
			gc_interval: opts.gc_interval,
			gc_full_scan_frequency: opts.gc_full_scan_frequency,
			cleanup_interval: opts.cleanup_interval,
		};
		// Start background tasks when enabled
		if opts.enable_cleanup {
			db.initialise_cleanup_worker();
		}
		if opts.enable_gc {
			db.initialise_garbage_worker();
		}
		// Return the database
		Ok(db)
	}

	/// Configure the database to use immediate garbage collection.
	///
	/// In this mode, old MVCC transaction entries are cleaned up
	/// during transaction commits as soon as they are no longer
	/// needed by any active read transactions. This ensures minimal
	/// memory usage while maintaining correctness for concurrent
	/// transactions. Additionally, if [`DatabaseOptions::enable_gc`]
	/// is set, a background thread will periodically clean up any
	/// stale versions.
	pub fn with_gc(self) -> Self {
		// Store the garbage collection epoch
		*self.garbage_collection_epoch.write() = None;
		// Return the database
		self
	}

	/// Configure the database to preserve versions for a specified duration.
	///
	/// In this mode, MVCC transaction entries are retained for at least
	/// the specified duration, allowing point-in-time reads within that
	/// window. Old versions are cleaned up during transaction commits
	/// once they exceed the history duration and are no longer needed
	/// by active transactions. Additionally, if [`DatabaseOptions::enable_gc`]
	/// is set, a background thread will periodically clean up stale
	/// versions across the entire datastore.
	pub fn with_gc_history(self, history: Duration) -> Self {
		// Store the garbage collection epoch
		*self.garbage_collection_epoch.write() = Some(history);
		// Return the database
		self
	}

	/// Start a new transaction on this database
	pub fn transaction(&self, write: bool) -> Transaction {
		self.pool.get(write)
	}

	/// Get a reference to the persistence layer if enabled
	#[cfg(not(target_arch = "wasm32"))]
	pub fn persistence(&self) -> Option<&Persistence> {
		self.persistence.as_ref()
	}

	/// Manually perform transaction queue cleanup.
	///
	/// This should be called when automatic cleanup is disabled via
	/// [`DatabaseOptions::enable_cleanup`].
	pub fn run_cleanup(&self) {
		// Use `u64::MAX` as the "no readers" fallback to preserve the
		// original semantics (trim nothing when nothing is registered).
		let oldest = self.earliest_active_commit(u64::MAX);
		if oldest != u64::MAX {
			self.transaction_commit_queue.range(..oldest).for_each(|e| {
				e.remove();
			});
		}
	}

	/// Manually perform garbage collection of stale record versions.
	///
	/// This function first processes keys that are known to have stale
	/// versions (tracked during transaction commits), then performs a full
	/// datastore scan to clean up any remaining old versions. Note that
	/// inline garbage collection happens automatically during transaction
	/// commits, but only for keys being modified. This function is useful
	/// for cleaning up stale versions on keys that haven't been recently
	/// modified, or when automatic background GC is disabled via
	/// [`DatabaseOptions::enable_gc`].
	pub fn run_gc(&self) {
		let cleanup_ts = self.compute_cleanup_ts();
		// First, process keys known to have stale versions
		self.run_gc_dirty_inner(cleanup_ts);
		// Then perform a full datastore scan for any remaining stale versions
		self.run_gc_full(cleanup_ts);
	}

	/// Process only dirty keys that are known to have stale versions.
	///
	/// This is a lightweight alternative to a full datastore scan, useful
	/// for frequent incremental cleanup between full GC passes.
	pub fn run_gc_dirty(&self) {
		let cleanup_ts = self.compute_cleanup_ts();
		// Process dirty keys
		self.run_gc_dirty_inner(cleanup_ts);
	}

	/// Compute the next `cleanup_ts`, publishing it into `gc_floor` so
	/// concurrent `register_counter` retries any reader whose snapshot
	/// is below it, then re-scan to bound by any newly-arrived reader.
	///
	/// The proposed value is capped at the current oracle timestamp.
	/// Without that cap, an idle database (oracle frozen while wall
	/// clock advances) would push `gc_floor` above any value a future
	/// reader could load — causing `register_counter` to spin forever
	/// retrying.
	fn compute_cleanup_ts(&self) -> u64 {
		let now = self.oracle.current_time_ns();
		let history = self.garbage_collection_epoch.read().unwrap_or_default().as_nanos();
		let history_cutoff = now.saturating_sub(history as u64);
		let earliest_tx = self.earliest_active_version(now);
		let oracle_now = self.oracle.inner.timestamp.load(Ordering::SeqCst);
		// `gc_floor` must stay <= oracle so any future reader's load
		// (which returns >= current oracle) satisfies the floor check.
		let proposed = history_cutoff.min(earliest_tx).min(oracle_now);
		// Publish proposed cleanup_ts via `gc_floor` BEFORE reclaiming.
		// A `register_counter` that fences after CAS-publish will see
		// either the old floor (and its publish becomes visible to our
		// re-scan below via fence-fence SC ordering) or the new floor
		// (and will retry to a fresh snapshot above it).
		self.gc_floor.fetch_max(proposed, Ordering::SeqCst);
		fence(Ordering::SeqCst);
		let earliest_after = self.earliest_active_version(now);
		proposed.min(earliest_after)
	}

	fn run_gc_dirty_inner(&self, cleanup_ts: u64) {
		// Drain all keys from the dirty queue
		let mut iter = self.datastore.raw_iter_mut();
		while let Some(key) = self.gc_dirty_keys.pop() {
			// Check if this key still exists in the datastore
			if iter.seek_exact(&key) {
				let (_, versions) = iter.next().expect("seek_exact returned true");
				// Clean up unnecessary older versions
				if versions.gc_older_versions(cleanup_ts) == 0 {
					// Remove the entry just yielded by next()
					iter.remove_here();
				}
			}
		}
	}

	fn run_gc_full(&self, cleanup_ts: u64) {
		// Iterate over the entire datastore
		let mut iter = self.datastore.raw_iter_mut();
		iter.seek_to_first();
		while let Some((_, versions)) = iter.next() {
			// Clean up unnecessary older versions
			if versions.gc_older_versions(cleanup_ts) == 0 {
				// Remove the entry just yielded by next()
				iter.remove_here();
			}
		}
	}

	/// Shutdown the datastore, waiting for background threads to exit
	fn shutdown(&self) {
		#[cfg(not(target_arch = "wasm32"))]
		{
			// First, disable Persistence background workers if present
			if let Some(ref persistence) = self.persistence {
				// Disable the persistence background workers
				persistence.background_threads_enabled.store(false, Ordering::Release);
				// Wait for persistence threads to exit
				if let Some(handle) = persistence.fsync_handle.write().take() {
					handle.thread().unpark();
					let _ = handle.join();
				}
				if let Some(handle) = persistence.snapshot_handle.write().take() {
					handle.thread().unpark();
					let _ = handle.join();
				}
				if let Some(handle) = persistence.appender_handle.write().take() {
					handle.thread().unpark();
					let _ = handle.join();
				}
			}
		}
		// Then disable Database background workers
		self.background_threads_enabled.store(false, Ordering::Relaxed);
		#[cfg(not(target_arch = "wasm32"))]
		{
			// Wait for the transaction cleanup thread to exit
			if let Some(handle) = self.transaction_cleanup_handle.write().take() {
				handle.thread().unpark();
				let _ = handle.join();
			}
			// Wait for the garbage collector thread to exit
			if let Some(handle) = self.garbage_collection_handle.write().take() {
				handle.thread().unpark();
				let _ = handle.join();
			}
		}
	}

	/// Start the transaction commit queue cleanup thread after creating the
	/// database
	#[cfg(not(target_arch = "wasm32"))]
	fn initialise_cleanup_worker(&self) {
		// Clone the underlying datastore inner
		let db = self.inner.clone();
		// Check if a background thread is already running
		if db.transaction_cleanup_handle.read().is_none() {
			// Get the specified interval
			let interval = self.cleanup_interval;
			// Spawn a new thread to handle periodic cleanup
			let handle = std::thread::spawn(move || {
				// Check whether the garbage collection process is enabled
				while db.background_threads_enabled.load(Ordering::Relaxed) {
					// Wait for a specified time interval
					std::thread::park_timeout(interval);
					// Check shutdown flag again after waking
					if !db.background_threads_enabled.load(Ordering::Relaxed) {
						break;
					}
					// Clean up the transaction commit queue
					let oldest = db.earliest_active_commit(u64::MAX);
					// Ensure the oldest value is not the max
					if oldest != u64::MAX {
						db.transaction_commit_queue.range(..oldest).for_each(|e| {
							e.remove();
						});
					}
				}
			});
			// Store and track the thread handle
			*self.inner.transaction_cleanup_handle.write() = Some(handle);
		}
	}

	/// Start the garbage collection thread after creating the database
	#[cfg(not(target_arch = "wasm32"))]
	fn initialise_garbage_worker(&self) {
		// Clone the underlying datastore inner
		let db = self.inner.clone();
		// Check if a background thread is already running
		if db.garbage_collection_handle.read().is_none() {
			// Get the specified interval
			let interval = self.gc_interval;
			// Full scan runs every Nth wake cycle; dirty-key pass runs every cycle
			let full_scan_frequency = self.gc_full_scan_frequency.max(1);
			// Spawn a new thread to handle periodic garbage collection
			let handle = std::thread::spawn(move || {
				let mut cycle: u64 = 0;
				// Check whether the garbage collection process is enabled
				while db.background_threads_enabled.load(Ordering::Relaxed) {
					// Wait for a specified time interval
					std::thread::park_timeout(interval);
					// Check shutdown flag again after waking
					if !db.background_threads_enabled.load(Ordering::Relaxed) {
						break;
					}
					// Compute the next cleanup_ts. This publishes the
					// proposed value to `gc_floor` and re-scans the
					// counter map so any reader landing in the
					// publish-and-scan window is either visible to us
					// (and bounds the final cleanup_ts) or retries from
					// a fresh oracle read above the floor. The proposed
					// value is capped at the current oracle so an idle
					// database does not lock readers out forever.
					let cleanup_ts = {
						let now = db.oracle.current_time_ns();
						let history =
							db.garbage_collection_epoch.read().unwrap_or_default().as_nanos();
						let history_cutoff = now.saturating_sub(history as u64);
						let earliest_tx = db.earliest_active_version(now);
						let oracle_now = db.oracle.inner.timestamp.load(Ordering::SeqCst);
						let proposed = history_cutoff.min(earliest_tx).min(oracle_now);
						db.gc_floor.fetch_max(proposed, Ordering::SeqCst);
						fence(Ordering::SeqCst);
						let earliest_after = db.earliest_active_version(now);
						proposed.min(earliest_after)
					};
					// Drain all keys from the dirty queue (incremental GC).
					// A fresh `raw_iter_mut` is opened per key so the
					// exclusive leaf guard is scoped to a single key —
					// this avoids any iterator-state bugs from reusing the
					// same cursor across `remove_here` calls that may
					// trigger leaf merges.
					while let Some(key) = db.gc_dirty_keys.pop() {
						let mut iter = db.datastore.raw_iter_mut();
						if iter.seek_exact(&key) {
							let (_, versions) = iter.next().expect("seek_exact returned true");
							if versions.gc_older_versions(cleanup_ts) == 0 {
								iter.remove_here();
							}
						}
					}
					// Periodically do a full datastore scan
					cycle += 1;
					if cycle.is_multiple_of(full_scan_frequency) {
						// Iterate over the entire datastore
						let mut iter = db.datastore.raw_iter_mut();
						iter.seek_to_first();
						while let Some((_, versions)) = iter.next() {
							// Clean up unnecessary older versions
							if versions.gc_older_versions(cleanup_ts) == 0 {
								// Remove the entry just yielded by next()
								iter.remove_here();
							}
						}
					}
				}
			});
			// Store and track the thread handle
			*self.inner.garbage_collection_handle.write() = Some(handle);
		}
	}
}

#[cfg(test)]
mod tests {

	use super::*;

	#[test]
	fn begin_tx() {
		let db = Database::new();
		db.transaction(false);
	}

	#[test]
	fn finished_tx_not_writeable() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		let res = tx.cancel();
		assert!(res.is_ok());
		let res = tx.put("test", "something");
		assert!(res.is_err());
		let res = tx.set("test", "something");
		assert!(res.is_err());
		let res = tx.del("test");
		assert!(res.is_err());
		let res = tx.commit();
		assert!(res.is_err());
		let res = tx.cancel();
		assert!(res.is_err());
	}

	#[test]
	fn cancelled_tx_is_cancelled() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("test", "something").unwrap();
		let res = tx.exists("test").unwrap();
		assert!(res);
		let res = tx.get("test").unwrap();
		assert_eq!(res.as_deref(), Some(b"something" as &[u8]));
		let res = tx.cancel();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.exists("test").unwrap();
		assert!(!res);
		let res = tx.get("test").unwrap();
		assert_eq!(res, None);
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn committed_tx_is_committed() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("test", "something").unwrap();
		let res = tx.exists("test").unwrap();
		assert!(res);
		let res = tx.get("test").unwrap();
		assert_eq!(res.as_deref(), Some(b"something" as &[u8]));
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.exists("test").unwrap();
		assert!(res);
		let res = tx.get("test").unwrap();
		assert_eq!(res.as_deref(), Some(b"something" as &[u8]));
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn multiple_concurrent_readers() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("test", "something").unwrap();
		let res = tx.exists("test").unwrap();
		assert!(res);
		let res = tx.get("test").unwrap();
		assert_eq!(res.as_deref(), Some(b"something" as &[u8]));
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx1 = db.transaction(false);
		let res = tx1.exists("test").unwrap();
		assert!(res);
		let res = tx1.exists("temp").unwrap();
		assert!(!res);
		// ----------
		let mut tx2 = db.transaction(false);
		let res = tx2.exists("test").unwrap();
		assert!(res);
		let res = tx2.exists("temp").unwrap();
		assert!(!res);
		// ----------
		let res = tx1.cancel();
		assert!(res.is_ok());
		let res = tx2.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn multiple_concurrent_operators() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("test", "something").unwrap();
		let res = tx.exists("test").unwrap();
		assert!(res);
		let res = tx.get("test").unwrap();
		assert_eq!(res.as_deref(), Some(b"something" as &[u8]));
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx1 = db.transaction(false);
		let res = tx1.exists("test").unwrap();
		assert!(res);
		let res = tx1.exists("temp").unwrap();
		assert!(!res);
		// ----------
		let mut txw = db.transaction(true);
		txw.put("temp", "other").unwrap();
		let res = txw.exists("test").unwrap();
		assert!(res);
		let res = txw.exists("temp").unwrap();
		assert!(res);
		let res = txw.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx2 = db.transaction(false);
		let res = tx2.exists("test").unwrap();
		assert!(res);
		let res = tx2.exists("temp").unwrap();
		assert!(res);
		// ----------
		let res = tx1.exists("temp").unwrap();
		assert!(!res);
		// ----------
		let res = tx1.cancel();
		assert!(res.is_ok());
		let res = tx2.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn iterate_keys_forward() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "a").unwrap();
		tx.put("b", "b").unwrap();
		tx.put("c", "c").unwrap();
		tx.put("d", "d").unwrap();
		tx.put("e", "e").unwrap();
		tx.put("f", "f").unwrap();
		tx.put("g", "g").unwrap();
		tx.put("h", "h").unwrap();
		tx.put("i", "i").unwrap();
		tx.put("j", "j").unwrap();
		tx.put("k", "k").unwrap();
		tx.put("l", "l").unwrap();
		tx.put("m", "m").unwrap();
		tx.put("n", "n").unwrap();
		tx.put("o", "o").unwrap();
		let res = tx.keys("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].as_ref(), b"c");
		assert_eq!(res[1], "d");
		assert_eq!(res[2], "e");
		assert_eq!(res[3], "f");
		assert_eq!(res[4], "g");
		assert_eq!(res[5], "h");
		assert_eq!(res[6], "i");
		assert_eq!(res[7], "j");
		assert_eq!(res[8], "k");
		assert_eq!(res[9], "l");
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.keys("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].as_ref(), b"c");
		assert_eq!(res[1], "d");
		assert_eq!(res[2], "e");
		assert_eq!(res[3], "f");
		assert_eq!(res[4], "g");
		assert_eq!(res[5], "h");
		assert_eq!(res[6], "i");
		assert_eq!(res[7], "j");
		assert_eq!(res[8], "k");
		assert_eq!(res[9], "l");
		let res = tx.cancel();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.keys("c".."z", Some(3), Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0], "f");
		assert_eq!(res[1], "g");
		assert_eq!(res[2], "h");
		assert_eq!(res[3], "i");
		assert_eq!(res[4], "j");
		assert_eq!(res[5], "k");
		assert_eq!(res[6], "l");
		assert_eq!(res[7], "m");
		assert_eq!(res[8], "n");
		assert_eq!(res[9], "o");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn iterate_keys_reverse() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "a").unwrap();
		tx.put("b", "b").unwrap();
		tx.put("c", "c").unwrap();
		tx.put("d", "d").unwrap();
		tx.put("e", "e").unwrap();
		tx.put("f", "f").unwrap();
		tx.put("g", "g").unwrap();
		tx.put("h", "h").unwrap();
		tx.put("i", "i").unwrap();
		tx.put("j", "j").unwrap();
		tx.put("k", "k").unwrap();
		tx.put("l", "l").unwrap();
		tx.put("m", "m").unwrap();
		tx.put("n", "n").unwrap();
		tx.put("o", "o").unwrap();
		let res = tx.keys_reverse("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0], "o");
		assert_eq!(res[1], "n");
		assert_eq!(res[2], "m");
		assert_eq!(res[3], "l");
		assert_eq!(res[4], "k");
		assert_eq!(res[5], "j");
		assert_eq!(res[6], "i");
		assert_eq!(res[7], "h");
		assert_eq!(res[8], "g");
		assert_eq!(res[9], "f");
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.keys_reverse("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0], "o");
		assert_eq!(res[1], "n");
		assert_eq!(res[2], "m");
		assert_eq!(res[3], "l");
		assert_eq!(res[4], "k");
		assert_eq!(res[5], "j");
		assert_eq!(res[6], "i");
		assert_eq!(res[7], "h");
		assert_eq!(res[8], "g");
		assert_eq!(res[9], "f");
		let res = tx.cancel();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.keys_reverse("c".."z", Some(3), Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0], "l");
		assert_eq!(res[1], "k");
		assert_eq!(res[2], "j");
		assert_eq!(res[3], "i");
		assert_eq!(res[4], "h");
		assert_eq!(res[5], "g");
		assert_eq!(res[6], "f");
		assert_eq!(res[7], "e");
		assert_eq!(res[8], "d");
		assert_eq!(res[9], "c");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn iterate_keys_values_forward() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "a").unwrap();
		tx.put("b", "b").unwrap();
		tx.put("c", "c").unwrap();
		tx.put("d", "d").unwrap();
		tx.put("e", "e").unwrap();
		tx.put("f", "f").unwrap();
		tx.put("g", "g").unwrap();
		tx.put("h", "h").unwrap();
		tx.put("i", "i").unwrap();
		tx.put("j", "j").unwrap();
		tx.put("k", "k").unwrap();
		tx.put("l", "l").unwrap();
		tx.put("m", "m").unwrap();
		tx.put("n", "n").unwrap();
		tx.put("o", "o").unwrap();
		let res = tx.scan("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].0.as_ref(), b"c");
		assert_eq!(res[0].1.as_ref(), b"c");
		assert_eq!(res[1].0.as_ref(), b"d");
		assert_eq!(res[1].1.as_ref(), b"d");
		assert_eq!(res[2].0.as_ref(), b"e");
		assert_eq!(res[3].0.as_ref(), b"f");
		assert_eq!(res[4].0.as_ref(), b"g");
		assert_eq!(res[5].0.as_ref(), b"h");
		assert_eq!(res[6].0.as_ref(), b"i");
		assert_eq!(res[7].0.as_ref(), b"j");
		assert_eq!(res[8].0.as_ref(), b"k");
		assert_eq!(res[9].0.as_ref(), b"l");
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.scan("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].0.as_ref(), b"c");
		assert_eq!(res[0].1.as_ref(), b"c");
		assert_eq!(res[1].0.as_ref(), b"d");
		assert_eq!(res[1].1.as_ref(), b"d");
		assert_eq!(res[2].0.as_ref(), b"e");
		assert_eq!(res[3].0.as_ref(), b"f");
		assert_eq!(res[4].0.as_ref(), b"g");
		assert_eq!(res[5].0.as_ref(), b"h");
		assert_eq!(res[6].0.as_ref(), b"i");
		assert_eq!(res[7].0.as_ref(), b"j");
		assert_eq!(res[8].0.as_ref(), b"k");
		assert_eq!(res[9].0.as_ref(), b"l");
		let res = tx.cancel();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.scan("c".."z", Some(3), Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].0.as_ref(), b"f");
		assert_eq!(res[1].0.as_ref(), b"g");
		assert_eq!(res[2].0.as_ref(), b"h");
		assert_eq!(res[3].0.as_ref(), b"i");
		assert_eq!(res[4].0.as_ref(), b"j");
		assert_eq!(res[5].0.as_ref(), b"k");
		assert_eq!(res[6].0.as_ref(), b"l");
		assert_eq!(res[7].0.as_ref(), b"m");
		assert_eq!(res[8].0.as_ref(), b"n");
		assert_eq!(res[9].0.as_ref(), b"o");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn iterate_keys_values_reverse() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "a").unwrap();
		tx.put("b", "b").unwrap();
		tx.put("c", "c").unwrap();
		tx.put("d", "d").unwrap();
		tx.put("e", "e").unwrap();
		tx.put("f", "f").unwrap();
		tx.put("g", "g").unwrap();
		tx.put("h", "h").unwrap();
		tx.put("i", "i").unwrap();
		tx.put("j", "j").unwrap();
		tx.put("k", "k").unwrap();
		tx.put("l", "l").unwrap();
		tx.put("m", "m").unwrap();
		tx.put("n", "n").unwrap();
		tx.put("o", "o").unwrap();
		let res = tx.scan_reverse("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].0.as_ref(), b"o");
		assert_eq!(res[1].0.as_ref(), b"n");
		assert_eq!(res[2].0.as_ref(), b"m");
		assert_eq!(res[3].0.as_ref(), b"l");
		assert_eq!(res[4].0.as_ref(), b"k");
		assert_eq!(res[5].0.as_ref(), b"j");
		assert_eq!(res[6].0.as_ref(), b"i");
		assert_eq!(res[7].0.as_ref(), b"h");
		assert_eq!(res[8].0.as_ref(), b"g");
		assert_eq!(res[9].0.as_ref(), b"f");
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.scan_reverse("c".."z", None, Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].0.as_ref(), b"o");
		assert_eq!(res[1].0.as_ref(), b"n");
		assert_eq!(res[2].0.as_ref(), b"m");
		assert_eq!(res[3].0.as_ref(), b"l");
		assert_eq!(res[4].0.as_ref(), b"k");
		assert_eq!(res[5].0.as_ref(), b"j");
		assert_eq!(res[6].0.as_ref(), b"i");
		assert_eq!(res[7].0.as_ref(), b"h");
		assert_eq!(res[8].0.as_ref(), b"g");
		assert_eq!(res[9].0.as_ref(), b"f");
		let res = tx.cancel();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.scan_reverse("c".."z", Some(3), Some(10)).unwrap();
		assert_eq!(res.len(), 10);
		assert_eq!(res[0].0.as_ref(), b"l");
		assert_eq!(res[1].0.as_ref(), b"k");
		assert_eq!(res[2].0.as_ref(), b"j");
		assert_eq!(res[3].0.as_ref(), b"i");
		assert_eq!(res[4].0.as_ref(), b"h");
		assert_eq!(res[5].0.as_ref(), b"g");
		assert_eq!(res[6].0.as_ref(), b"f");
		assert_eq!(res[7].0.as_ref(), b"e");
		assert_eq!(res[8].0.as_ref(), b"d");
		assert_eq!(res[9].0.as_ref(), b"c");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn count_keys_values() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "a").unwrap();
		tx.put("b", "b").unwrap();
		tx.put("c", "c").unwrap();
		tx.put("d", "d").unwrap();
		tx.put("e", "e").unwrap();
		tx.put("f", "f").unwrap();
		tx.put("g", "g").unwrap();
		tx.put("h", "h").unwrap();
		tx.put("i", "i").unwrap();
		tx.put("j", "j").unwrap();
		tx.put("k", "k").unwrap();
		tx.put("l", "l").unwrap();
		tx.put("m", "m").unwrap();
		tx.put("n", "n").unwrap();
		tx.put("o", "o").unwrap();
		let res = tx.total("c".."z", None, Some(10)).unwrap();
		assert_eq!(res, 10);
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let res = tx.total("c".."z", Some(3), Some(10)).unwrap();
		assert_eq!(res, 10);
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	// --------------------------------------------------
	// Cursor and Iterator Tests
	// --------------------------------------------------

	#[test]
	fn cursor_forward_iteration() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let mut cursor = tx.cursor("a".."z").unwrap();
		cursor.seek_to_first();
		// Check first entry
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"a");
		assert_eq!(cursor.value().unwrap().as_ref(), b"1");
		// Move forward
		cursor.next();
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"b");
		// Continue to end
		cursor.next(); // c
		cursor.next(); // d
		cursor.next(); // e
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"e");
		cursor.next();
		assert!(!cursor.valid());
		drop(cursor);
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn cursor_reverse_iteration() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let mut cursor = tx.cursor("a".."z").unwrap();
		cursor.seek_to_last();
		// Check last entry
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"e");
		assert_eq!(cursor.value().unwrap().as_ref(), b"5");
		// Move backward
		cursor.prev();
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"d");
		// Continue to beginning
		cursor.prev(); // c
		cursor.prev(); // b
		cursor.prev(); // a
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"a");
		cursor.prev();
		assert!(!cursor.valid());
		drop(cursor);
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn cursor_seek_operations() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("e", "5").unwrap();
		tx.put("g", "7").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let mut cursor = tx.cursor("a".."z").unwrap();
		// Seek to exact key
		cursor.seek("c");
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"c");
		// Seek to non-existent key (should land on next)
		cursor.seek("d");
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"e");
		// Seek for prev to exact key
		cursor.seek_for_prev("e");
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"c");
		// Seek beyond range
		cursor.seek("z");
		assert!(!cursor.valid());
		drop(cursor);
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn cursor_bidirectional_switch() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let mut cursor = tx.cursor("a".."z").unwrap();
		// Start forward
		cursor.seek_to_first();
		assert_eq!(cursor.key().unwrap().as_ref(), b"a");
		cursor.next();
		assert_eq!(cursor.key().unwrap().as_ref(), b"b");
		cursor.next();
		assert_eq!(cursor.key().unwrap().as_ref(), b"c");
		// Switch to reverse
		cursor.prev();
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"b");
		// Switch back to forward
		cursor.next();
		assert!(cursor.valid());
		assert_eq!(cursor.key().unwrap().as_ref(), b"c");
		drop(cursor);
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn keys_iterator_forward() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let keys: Vec<_> = tx.keys_iter("b".."e").unwrap().collect();
		assert_eq!(keys.len(), 3);
		assert_eq!(keys[0].as_ref(), b"b");
		assert_eq!(keys[1].as_ref(), b"c");
		assert_eq!(keys[2].as_ref(), b"d");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn keys_iterator_reverse() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let keys: Vec<_> = tx.keys_iter_reverse("b".."e").unwrap().collect();
		assert_eq!(keys.len(), 3);
		assert_eq!(keys[0].as_ref(), b"d");
		assert_eq!(keys[1].as_ref(), b"c");
		assert_eq!(keys[2].as_ref(), b"b");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn keys_iterator_with_take() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		tx.put("f", "6").unwrap();
		tx.put("g", "7").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let keys: Vec<_> = tx.keys_iter("a".."z").unwrap().take(3).collect();
		assert_eq!(keys.len(), 3);
		assert_eq!(keys[0].as_ref(), b"a");
		assert_eq!(keys[1].as_ref(), b"b");
		assert_eq!(keys[2].as_ref(), b"c");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn keys_iterator_with_skip() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let keys: Vec<_> = tx.keys_iter("a".."z").unwrap().skip(2).collect();
		assert_eq!(keys.len(), 3);
		assert_eq!(keys[0].as_ref(), b"c");
		assert_eq!(keys[1].as_ref(), b"d");
		assert_eq!(keys[2].as_ref(), b"e");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn scan_iterator_forward() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let pairs: Vec<_> = tx.scan_iter("a".."z").unwrap().collect();
		assert_eq!(pairs.len(), 3);
		assert_eq!(pairs[0].0.as_ref(), b"a");
		assert_eq!(pairs[0].1.as_ref(), b"1");
		assert_eq!(pairs[1].0.as_ref(), b"b");
		assert_eq!(pairs[1].1.as_ref(), b"2");
		assert_eq!(pairs[2].0.as_ref(), b"c");
		assert_eq!(pairs[2].1.as_ref(), b"3");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn scan_iterator_reverse() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let pairs: Vec<_> = tx.scan_iter_reverse("a".."z").unwrap().collect();
		assert_eq!(pairs.len(), 3);
		assert_eq!(pairs[0].0.as_ref(), b"c");
		assert_eq!(pairs[0].1.as_ref(), b"3");
		assert_eq!(pairs[1].0.as_ref(), b"b");
		assert_eq!(pairs[1].1.as_ref(), b"2");
		assert_eq!(pairs[2].0.as_ref(), b"a");
		assert_eq!(pairs[2].1.as_ref(), b"1");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn scan_iterator_with_take() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		tx.put("d", "4").unwrap();
		tx.put("e", "5").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(false);
		let pairs: Vec<_> = tx.scan_iter("a".."z").unwrap().take(2).collect();
		assert_eq!(pairs.len(), 2);
		assert_eq!(pairs[0].0.as_ref(), b"a");
		assert_eq!(pairs[1].0.as_ref(), b"b");
		let res = tx.cancel();
		assert!(res.is_ok());
	}

	#[test]
	fn iterator_sees_uncommitted_writes() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		// Before commit, iterator should see uncommitted writes
		let keys: Vec<_> = tx.keys_iter("a".."z").unwrap().collect();
		assert_eq!(keys.len(), 2);
		assert_eq!(keys[0].as_ref(), b"a");
		assert_eq!(keys[1].as_ref(), b"b");
		let res = tx.commit();
		assert!(res.is_ok());
	}

	#[test]
	fn cursor_handles_deleted_entries() {
		let db = Database::new();
		// ----------
		let mut tx = db.transaction(true);
		tx.put("a", "1").unwrap();
		tx.put("b", "2").unwrap();
		tx.put("c", "3").unwrap();
		let res = tx.commit();
		assert!(res.is_ok());
		// ----------
		let mut tx = db.transaction(true);
		tx.del("b").unwrap();
		// Iterator should skip deleted entry
		let keys: Vec<_> = tx.keys_iter("a".."z").unwrap().collect();
		assert_eq!(keys.len(), 2);
		assert_eq!(keys[0].as_ref(), b"a");
		assert_eq!(keys[1].as_ref(), b"c");
		let res = tx.cancel();
		assert!(res.is_ok());
	}
}