surrealdb-core 3.2.3

use std::any::{Any, TypeId};
#[cfg(not(target_family = "wasm"))]
use std::collections::HashMap;
#[cfg(target_family = "wasm")]
use std::collections::HashSet;
#[cfg(not(target_family = "wasm"))]
use std::collections::hash_map::Entry;
use std::fmt::{self, Display};
#[cfg(storage)]
use std::path::PathBuf;
use std::pin::pin;
use std::sync::Arc;
use std::task::{Poll, ready};
use std::time::Duration;

#[allow(unused_imports)]
use anyhow::bail;
use anyhow::{Context as _, Result, ensure};
use async_channel::Sender;
use bytes::{Bytes, BytesMut};
use futures::{Future, Stream};
use rand::Rng;
use reblessive::TreeStack;
use surrealdb_types::{AuthError, Error as TypesError, SurrealValue, object};
#[cfg(not(target_family = "wasm"))]
use tokio::spawn;
use tokio::sync::Notify;
use tokio::time::{Instant, sleep, timeout, timeout_at};
use tokio_util::sync::CancellationToken;
use tracing::{debug, instrument, trace, warn};
use uuid::Uuid;

use super::api::{BoxFut, Transactable};
use super::tr::Transactor;
use super::tx::Transaction;
use super::version::MajorVersion;
use super::{Key, Val, export};
use crate::api::err::ApiError;
use crate::api::invocation::process_api_request;
use crate::api::request::ApiRequest;
use crate::api::response::ApiResponse;
use crate::buc::manager::BucketsManager;
use crate::catalog::providers::{
	ApiProvider, CatalogProvider, DatabaseProvider, NamespaceProvider, NodeProvider, TableProvider,
	UserProvider,
};
use crate::catalog::{ApiDefinition, Index, NodeLiveQuery, SubscriptionDefinition};
use crate::cnf::dynamic::DynamicConfiguration;
use crate::cnf::{CommonConfig, ConfigMap, LiveQueryEngine};
use crate::ctx::{CancelHandle, Context};
#[cfg(feature = "jwks")]
use crate::dbs::capabilities::NetTarget;
use crate::dbs::capabilities::{
	ArbitraryQueryTarget, EvalQueryTarget, ExperimentalTarget, MethodTarget, RouteTarget,
};
use crate::dbs::node::{Node, Timestamp};
use crate::dbs::{
	Capabilities, DurableSession, Executor, MessageBroker, Options, QueryResult,
	QueryResultBuilder, Session,
};
use crate::doc::AsyncEventRecord;
use crate::err::Error;
use crate::exec::function::FunctionRegistry;
use crate::expr::model::get_model_path;
use crate::expr::statements::{DefineModelStatement, DefineStatement, DefineUserStatement};
use crate::expr::{Base, Expr, FlowResultExt as _, Literal, LogicalPlan, TopLevelExpr};
#[cfg(feature = "gql")]
use crate::gql::PreparedGqlQuery;
#[cfg(feature = "http")]
use crate::http::HttpClient;
use crate::iam::{Action, Auth, Error as IamError, Resource, ResourceKind, Role};
use crate::idx::IndexKeyBase;
use crate::idx::index::IndexOperation;
use crate::idx::trees::store::IndexStores;
use crate::key::root::ic::IndexCompactionKey;
use crate::key::root::rc::{
	RECLAIM_DATABASE, RECLAIM_INDEX, RECLAIM_NAMESPACE, ReclaimKey, ReclaimState,
};
use crate::kvs::LockType::*;
use crate::kvs::TransactionType::*;
use crate::kvs::cache::ds::DatastoreCache;
use crate::kvs::clock::SystemClock;
use crate::kvs::ds::requirements::{
	TransactionBuilderFactoryRequirements, TransactionBuilderRequirements,
};
use crate::kvs::index::IndexBuilder;
use crate::kvs::sequences::Sequences;
use crate::kvs::slowlog::SlowLog;
use crate::kvs::tasklease::{LeaseHandler, TaskLeaseType};
#[cfg(test)]
use crate::kvs::testing::{RetryableConflictSite, maybe_inject_retryable_conflict};
use crate::kvs::{
	Error as KvsError, KVValue, LockType, NORMAL_BATCH_SIZE, TransactionType,
	is_retryable_transaction_conflict,
};
use crate::lq::LiveQueryRouter;
use crate::observe::{ExecutionObserver, NoopObserver};
use crate::sql::Ast;
#[cfg(feature = "surrealism")]
use crate::surrealism::cache::SurrealismCache;
use crate::syn::parser::{ParserSettings, StatementStream};
use crate::types::{PublicNotification, PublicValue, PublicVariables};
use crate::val::convert_value_to_public_value;
use crate::{CommunityComposer, syn};

mod builder;
pub use builder::Builder;

const TARGET: &str = "surrealdb::core::kvs::ds";
const NODE_DELETE_TIMEOUT: Duration = Duration::from_secs(60);

/// The role assigned to the initial user created when starting the server with
/// credentials for the first time
const INITIAL_USER_ROLE: &str = "owner";

/// Whether a conditional write (`put_compare_key` / `del_compare_key`) failed
/// because its condition was not met — the key already existed, was deleted,
/// or changed since the guard value was read. On last-writer-wins backends
/// (TiKV) the condition is validated at commit, so this can surface from the
/// conditional call itself or from the subsequent `commit`; callers must check
/// both. This is how the durable RPC session writes stay atomic on TiKV, where
/// a blind `set`/`clr` is last-writer-wins (see the `multiwriter_same_keys_*`
/// KV coverage).
///
/// The error arrives either as a bare [`KvsError`] (e.g. from `commit()`, which
/// returns the backend error directly) or wrapped as [`Error::Kvs`] (from the
/// transaction helpers' `map_err(Error::from)`), so both forms are checked —
/// mirroring [`is_retryable_transaction_conflict`].
fn is_conditional_write_conflict(err: &anyhow::Error) -> bool {
	fn is_condition_error(e: &KvsError) -> bool {
		matches!(e, KvsError::TransactionConditionNotMet | KvsError::TransactionKeyAlreadyExists)
	}
	if let Some(e) = err.downcast_ref::<KvsError>() {
		return is_condition_error(e);
	}
	matches!(err.downcast_ref::<Error>(), Some(Error::Kvs(e)) if is_condition_error(e))
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ShutdownNodeDeleteOutcome {
	Archived,
	Failed,
	TimedOut,
}

async fn await_node_step<T, Fut>(
	deadline: Instant,
	timeout_duration: Duration,
	canceller: Option<&CancellationToken>,
	step: Fut,
) -> Result<T>
where
	Fut: Future<Output = Result<T>>,
{
	if let Some(canceller) = canceller {
		tokio::select! {
			biased;
			_ = canceller.cancelled() => bail!(Error::QueryCancelled),
			result = timeout_at(deadline, step) => match result {
				Ok(result) => result,
				Err(_) => bail!(Error::QueryTimedout(timeout_duration.into())),
			},
		}
	} else {
		match timeout_at(deadline, step).await {
			Ok(result) => result,
			Err(_) => bail!(Error::QueryTimedout(timeout_duration.into())),
		}
	}
}

async fn await_node_tx_step<T, Fut>(
	txn: &Transaction,
	deadline: Instant,
	timeout_duration: Duration,
	canceller: Option<&CancellationToken>,
	step: Fut,
) -> Result<T>
where
	Fut: Future<Output = Result<T>>,
{
	let result = if let Some(canceller) = canceller {
		tokio::select! {
			biased;
			_ = canceller.cancelled() => {
				let _ = txn.cancel().await;
				bail!(Error::QueryCancelled);
			}
			result = timeout_at(deadline, step) => result,
		}
	} else {
		timeout_at(deadline, step).await
	};

	match result {
		Ok(Ok(value)) => Ok(value),
		Ok(Err(e)) => {
			let _ = txn.cancel().await;
			Err(e)
		}
		Err(_) => {
			let _ = txn.cancel().await;
			bail!(Error::QueryTimedout(timeout_duration.into()))
		}
	}
}

fn archive_node_for_shutdown(
	timeout_duration: Duration,
	result: Result<()>,
) -> ShutdownNodeDeleteOutcome {
	match result {
		Ok(()) => ShutdownNodeDeleteOutcome::Archived,
		Err(e) => {
			if matches!(e.downcast_ref::<Error>(), Some(Error::QueryTimedout(_))) {
				warn!(
					target: TARGET,
					timeout = ?timeout_duration,
					"Timed out archiving node during shutdown; continuing shutdown"
				);
				return ShutdownNodeDeleteOutcome::TimedOut;
			}

			warn!(
				target: TARGET,
				error = %e,
				"Failed to archive node during shutdown; continuing shutdown"
			);
			ShutdownNodeDeleteOutcome::Failed
		}
	}
}

/// The underlying datastore instance which stores the dataset.
pub struct Datastore {
	transaction_factory: TransactionFactory,
	/// The unique id of this datastore, used in notifications.
	id: Uuid,
	/// Whether authentication is enabled on this datastore.
	auth_enabled: bool,
	/// The maximum duration timeout for running multiple statements in a query.
	dynamic_configuration: DynamicConfiguration,
	/// The slow log configuration determining when a query should be logged
	slow_log: Option<SlowLog>,
	/// The maximum duration timeout for running multiple statements in a
	/// transaction.
	transaction_timeout: Option<Duration>,
	/// The security and feature capabilities for this datastore.
	capabilities: Arc<Capabilities>,
	/// Broker used to deliver live-query notifications after their write commits.
	///
	/// `Some` iff live-query subscribers exist for this datastore (the broker owns the sender
	/// half of the notification channel internally). `None` disables live-query work entirely
	/// at the executor boundary.
	live_query_broker: Option<Arc<dyn MessageBroker>>,
	/// Per-node live-query router state (the tail cursor over the `lqe`
	/// keyspace). Only used when `config.live_query_engine` is `Router`, where
	/// the router is the sole notification delivery path. See [`crate::lq`].
	live_query_router: Arc<LiveQueryRouter>,
	/// Public HTTP endpoint this datastore publishes on its `Node` catalog row so other
	/// cluster members can route cross-node messages (e.g. live-query relay) to it.
	/// `None` in deployments that don't expose such an endpoint.
	http_endpoint: Option<String>,
	// The index store cache
	index_stores: IndexStores,
	// The cross transaction cache
	cache: Arc<DatastoreCache>,
	/// Cache of generated GraphQL schemas, shared across every transport that
	/// serves GraphQL: the HTTP `/graphql` route, GraphQL-over-WebSocket
	/// subscriptions, the `graphql` RPC method, and the MCP `graphql` tool.
	/// Keyed by `(ns, db, config, schema-fingerprint)` so DDL changes invalidate
	/// it automatically.
	#[cfg(all(feature = "graphql", not(target_family = "wasm")))]
	graphql_schema_cache: crate::graphql::cache::GraphQLSchemaCache,
	/// Registry of built-in scalar, aggregate, projection and index
	/// functions, along with the method-dispatch table. Built once when the
	/// datastore is constructed and shared across all transactions via the
	/// `Arc`. Every `Context` clones this `Arc` rather than rebuilding the
	/// registry, which is otherwise the single biggest per-query cost.
	function_registry: Arc<FunctionRegistry>,
	// The index asynchronous builder
	index_builder: IndexBuilder,
	#[cfg(storage)]
	// The temporary directory
	temporary_directory: Option<Arc<PathBuf>>,
	// Map of bucket connections
	buckets: BucketsManager,
	// The sequences
	sequences: Sequences,
	// The surrealism cache
	#[cfg(feature = "surrealism")]
	surrealism_cache: Arc<SurrealismCache>,
	/// When `true`, surrealism modules are loaded lazily on first use
	/// instead of being eagerly compiled at startup.
	#[cfg(feature = "surrealism")]
	lazy_surrealism: bool,
	// Async event processing trigger
	async_event_trigger: Arc<Notify>,
	/// Config
	config: Arc<CommonConfig>,
	// Http client used to make requests.
	#[cfg(feature = "http")]
	http_client: Arc<HttpClient>,
	/// Observer invoked on significant events. Defaults to [`NoopObserver`].
	observer: Arc<dyn ExecutionObserver>,
}

/// Represents a collection of metrics for a specific datastore flavor.
///
/// This structure is used to expose datastore-specific metrics to the telemetry system.
pub struct Metrics {
	/// The name of the metrics group (e.g., "surrealdb.rocksdb").
	pub name: &'static str,
	/// A list of u64-based metrics.
	pub u64_metrics: Vec<Metric>,
}

/// Represents a single metric with a name and description.
pub struct Metric {
	/// The name of the metric.
	pub name: &'static str,
	/// A human-readable description of the metric.
	pub description: &'static str,
}

#[derive(Clone)]
pub(crate) struct TransactionFactory {
	// The inner datastore type
	builder: Arc<Box<dyn TransactionBuilder>>,
	// Async event processing trigger
	async_event_trigger: Arc<Notify>,
	/// Observer invoked on transaction lifecycle events. Defaults to
	/// [`NoopObserver`]; replaced by the datastore's observer when one is
	/// configured.
	observer: Arc<dyn ExecutionObserver>,
	config: Arc<CommonConfig>,
}

impl TransactionFactory {
	pub(super) fn new(
		async_event_trigger: Arc<Notify>,
		builder: Box<dyn TransactionBuilder>,
		config: Arc<CommonConfig>,
	) -> Self {
		Self {
			builder: Arc::new(builder),
			async_event_trigger,
			observer: Arc::new(NoopObserver),
			config,
		}
	}

	/// Replace the observer. Used by the datastore builder to propagate the
	/// chosen observer to all transactions created after the swap.
	pub(crate) fn with_observer(mut self, observer: Arc<dyn ExecutionObserver>) -> Self {
		self.observer = observer;
		self
	}

	/// Access the observer. Transaction instrumentation fires events through
	/// this handle.
	#[allow(dead_code)]
	pub(crate) fn observer(&self) -> &Arc<dyn ExecutionObserver> {
		&self.observer
	}

	#[allow(
		unreachable_code,
		unreachable_patterns,
		unused_variables,
		reason = "Some variables are unused when no backends are enabled."
	)]
	pub async fn transaction(
		&self,
		write: TransactionType,
		lock: LockType,
		sequences: Sequences,
	) -> Result<Transaction> {
		// Specify if the transaction is writeable
		let write = match write {
			Read => false,
			Write => true,
		};
		// Specify if the transaction is lockable
		let lock = match lock {
			Pessimistic => true,
			Optimistic => false,
		};
		// Create a new transaction on the datastore
		let (inner, local) = self.builder.new_transaction(write, lock).await?;
		Ok(Transaction::new(
			local,
			sequences,
			Arc::clone(&self.async_event_trigger),
			Arc::clone(&self.observer),
			Transactor {
				inner,
			},
			&self.config,
		))
	}

	/// Registers metrics for the current datastore flavor if supported.
	fn register_metrics(&self) -> Option<Metrics> {
		self.builder.register_metrics()
	}

	/// Collects a specific u64 metric by name if supported by the datastore flavor.
	fn collect_u64_metric(&self, metric: &str) -> Option<u64> {
		self.builder.collect_u64_metric(metric)
	}
}

/// Abstraction over storage backends for creating and managing transactions.
///
/// This trait allows decoupling `Datastore` from concrete KV engines (memory,
/// RocksDB, TiKV, SurrealKV, SurrealDS, etc.). Implementors translate the
/// generic transaction parameters into a backend-specific transaction and
/// report whether the transaction is considered "local" (used internally to
/// enable some optimizations).
///
/// This was introduced to make the server more composable/embeddable. External
/// crates can implement `TransactionBuilder` to plug in custom backends while
/// reusing the rest of SurrealDB.
pub trait TransactionBuilder: TransactionBuilderRequirements {
	/// Create a new backend transaction.
	///
	/// - `write`: whether the transaction is writable (Write vs Read)
	/// - `lock`: whether pessimistic locking is requested
	///
	/// Returns the backend transaction object and a flag indicating if the
	/// transaction is local to the process (true) or requires external resources
	/// (false).
	fn new_transaction(
		&self,
		write: bool,
		lock: bool,
	) -> BoxFut<'_, Result<(Box<dyn Transactable>, bool)>>;

	/// Perform any backend-specific shutdown/cleanup.
	fn shutdown(&self) -> BoxFut<'_, Result<()>>;

	/// Registers metrics for the current datastore flavor if supported.
	///
	/// This will return a list of available metrics and their descriptions.
	fn register_metrics(&self) -> Option<Metrics>;

	/// Collects a specific u64 metric by name if supported by the datastore flavor.
	///
	/// - `metric`: The name of the metric to collect.
	fn collect_u64_metric(&self, metric: &str) -> Option<u64>;

	/// Returns an immutable backend-specific extension handle.
	///
	/// Backends expose only stable, shareable handles through this hook. The
	/// default implementation keeps community datastores free of extension
	/// state.
	///
	/// This is the extension point for backend-specific operations that
	/// don't fit the generic transaction interface: e.g. the TiKV backend
	/// returns its [`crate::kvs::tikv::TikvOpsHandle`] (matched on
	/// `TypeId`) so the engine can offer MVCC-GC / lock-cleanup /
	/// `unsafe_destroy_range` to operators without polluting this trait
	/// with TiKV-only signatures every other backend would have to no-op.
	fn extension(&self, _: TypeId) -> Option<Arc<dyn Any + Send + Sync>> {
		None
	}
}

/// Transaction-builder construction result with router startup state.
///
/// The datastore consumes `builder`; server startup threads `router_state` into
/// the router factory so embedders can make immutable handles available to
/// their HTTP routes without process globals.
pub struct TransactionBuilderParts<S> {
	/// Transaction builder consumed by the datastore.
	pub builder: Box<dyn TransactionBuilder>,
	/// Immutable router startup state produced by the composer.
	pub router_state: S,
}

impl<S> TransactionBuilderParts<S> {
	/// Construct transaction-builder parts with router startup state.
	pub fn new(builder: Box<dyn TransactionBuilder>, router_state: S) -> Self {
		Self {
			builder,
			router_state,
		}
	}
}

impl TransactionBuilderParts<()> {
	/// Construct transaction-builder parts for composers without router state.
	pub fn without_router_state(builder: Box<dyn TransactionBuilder>) -> Self {
		Self::new(builder, ())
	}
}

/// Factory that parses a datastore path and returns a concrete `TransactionBuilder`.
///
/// Implementations can decide how to interpret connection strings (e.g. "memory",
/// "rocksdb:...", "tikv:...") and which clock to use. This lets the CLI and
/// server be generic over different storage backends without hard-coding them.
///
/// The `path_valid` helper is used by the CLI to validate the path early and
/// provide better error messages before starting the runtime.
pub trait TransactionBuilderFactory: TransactionBuilderFactoryRequirements {
	/// Immutable state threaded into router construction after datastore startup.
	type RouterState: Clone + Send + Sync + 'static;

	/// Create a new transaction builder for the datastore.
	///
	/// # Parameters
	/// - `path`: Database connection path string
	/// - `canceller`: Token for graceful shutdown and cancellation of long-running operations
	#[cfg(not(target_family = "wasm"))]
	fn new_transaction_builder(
		&self,
		path: &str,
		canceller: CancellationToken,
		config: ConfigMap,
	) -> impl Future<Output = Result<TransactionBuilderParts<Self::RouterState>>> + Send;

	/// Create a new transaction builder for the datastore (WASM: no `Send` bound on the future).
	#[cfg(target_family = "wasm")]
	fn new_transaction_builder(
		&self,
		path: &str,
		canceller: CancellationToken,
		config: ConfigMap,
	) -> impl Future<Output = Result<TransactionBuilderParts<Self::RouterState>>>;

	/// Validate a datastore path string.
	fn path_valid(v: &str) -> Result<String>;

	/// Returns the stable datastore node id used for live-query ownership metadata.
	///
	/// Composers that run SurrealDB inside a clustered product should return a deterministic
	/// value so remote writers can route notifications back to the node that owns each
	/// subscriber connection.
	fn datastore_node_id(&self) -> Option<[u8; 16]> {
		None
	}

	/// Creates the broker that receives buffered live-query notifications after commit.
	///
	/// The default broker is local-only and preserves community behaviour. Clustered composers
	/// can return a broker that forwards remote targets without changing transaction results on
	/// delivery failure.
	fn live_query_broker(&self, channel: Sender<PublicNotification>) -> Arc<dyn MessageBroker> {
		crate::dbs::LocalMessageBroker::new(channel)
	}

	/// Public HTTP endpoint this datastore should publish for cross-node messaging.
	///
	/// Clustered composers surface their local node's endpoint here so the [`Datastore`]
	/// can record it on the `Node` catalog row, making it discoverable by peer nodes
	/// (e.g. for the live-query relay). Returns `None` in single-node and shared-backend
	/// deployments that don't expose a cross-node messaging endpoint.
	fn http_endpoint(&self) -> Option<String> {
		None
	}
}

pub mod requirements {
	use std::fmt::Display;

	#[cfg(target_family = "wasm")]
	pub trait TransactionBuilderRequirements: Display {}

	#[cfg(not(target_family = "wasm"))]
	pub trait TransactionBuilderRequirements: Display + Send + Sync + 'static {}

	#[cfg(target_family = "wasm")]
	pub trait TransactionBuilderFactoryRequirements {}

	#[cfg(not(target_family = "wasm"))]
	pub trait TransactionBuilderFactoryRequirements: Send + Sync + 'static {}
}

pub enum DatastoreFlavor {
	#[cfg(feature = "kv-mem")]
	Mem(super::mem::Datastore),
	#[cfg(feature = "kv-rocksdb")]
	RocksDB(super::rocksdb::Datastore),
	#[cfg(feature = "kv-indxdb")]
	IndxDB(super::indxdb::Datastore),
	#[cfg(feature = "kv-tikv")]
	TiKV(super::tikv::Datastore),
	#[cfg(feature = "kv-surrealkv")]
	SurrealKV(super::surrealkv::Datastore),
}

impl TransactionBuilderFactoryRequirements for CommunityComposer {}

impl TransactionBuilderFactory for CommunityComposer {
	type RouterState = ();

	#[allow(unused_variables)]
	async fn new_transaction_builder(
		&self,
		path: &str,
		_canceller: CancellationToken,
		config: ConfigMap,
	) -> Result<TransactionBuilderParts<Self::RouterState>> {
		// Extract query parameters from the path before scheme extraction
		let (raw_path, config_string) = match path.split_once('?') {
			Some((p, q)) => (p, Some(q)),
			None => (path, None),
		};

		let config = if let Some(config_string) = config_string {
			config.join(
				ConfigMap::from_config_string(config_string).map_keys(|x| format!("datastore_{x}")),
			)
		} else {
			config
		};

		// Extract the scheme and path components
		let (flavour, path) = match raw_path.split_once("://").or_else(|| raw_path.split_once(':'))
		{
			None if raw_path == "memory" => ("memory", ""),
			// Treat "mem" as an alias for "memory"
			None if raw_path == "mem" => ("memory", ""),
			Some(("mem", path)) => ("memory", path),
			Some((flavour, path)) => (flavour, path),
			// Validated already in the CLI, should never happen
			_ => bail!(Error::Unreachable("Provide a valid database path parameter".to_owned())),
		};

		let path = if path.starts_with("/") {
			// if absolute, remove all slashes except one
			let normalised = format!("/{}", path.trim_start_matches("/"));
			info!(target: TARGET, "Starting kvs store at absolute path {flavour}:{normalised}");
			normalised
		} else if path.is_empty() {
			info!(target: TARGET, "Starting kvs store in memory");
			"".to_string()
		} else {
			info!(target: TARGET, "Starting kvs store at relative path {flavour}://{path}");
			path.to_string()
		};
		// Initiate the desired datastore
		match (flavour, path) {
			// Initiate an in-memory datastore
			(flavour @ "memory", path) => {
				#[cfg(feature = "kv-mem")]
				{
					// Create a new blocking threadpool
					super::threadpool::initialise();

					// Persist path comes from the URL path; do not inject an empty
					// string or `parse_key_with` logs a spurious DATASTORE_PERSIST warning.
					let config = if path.is_empty() {
						config
					} else {
						config.with_key_value("datastore_persist", path)
					};
					// Parse SurrealMX configuration from URL path and query parameters
					let config = config.load();
					// Initialise the storage engine
					let v = super::mem::Datastore::new(config).await.map(DatastoreFlavor::Mem)?;
					info!(target: TARGET, "Started kvs store in {flavour}");
					Ok(TransactionBuilderParts::without_router_state(Box::<DatastoreFlavor>::new(
						v,
					)))
				}
				#[cfg(not(feature = "kv-mem"))]
				bail!(Error::Kvs(crate::kvs::Error::Datastore("Cannot connect to the `memory` storage engine as it is not enabled in this build of SurrealDB".to_owned())));
			}
			// The `file:` scheme has been removed. Catch it here so users
			// with legacy paths get a targeted message instead of the
			// generic fallback below.
			("file", _) => {
				bail!(Error::Kvs(crate::kvs::Error::Datastore(
					"The `file://` scheme is no longer supported; use `rocksdb://` or `surrealkv://` instead"
						.into()
				)));
			}
			// Initiate a RocksDB datastore
			(flavour @ "rocksdb", path) => {
				#[cfg(feature = "kv-rocksdb")]
				{
					// Create a new blocking threadpool
					super::threadpool::initialise();
					// Parse RocksDB-specific configuration from query parameters
					let config = config.load();
					// Initialise the storage engine
					let v = super::rocksdb::Datastore::new(&path, config)
						.await
						.map(DatastoreFlavor::RocksDB)?;
					info!(target: TARGET, "Started {flavour} kvs store");
					Ok(TransactionBuilderParts::without_router_state(Box::<DatastoreFlavor>::new(
						v,
					)))
				}
				#[cfg(not(feature = "kv-rocksdb"))]
				bail!(Error::Kvs(crate::kvs::Error::Datastore("Cannot connect to the `rocksdb` storage engine as it is not enabled in this build of SurrealDB".to_owned())));
			}
			// Initiate a SurrealKV database
			(flavour @ "surrealkv", path) => {
				#[cfg(feature = "kv-surrealkv")]
				{
					// Create a new blocking threadpool
					super::threadpool::initialise();
					// Parse SurrealKV-specific configuration from query parameters
					let config = config.load();
					// Initialise the storage engine
					let v = super::surrealkv::Datastore::new(&path, config)
						.await
						.map(DatastoreFlavor::SurrealKV)?;
					info!(target: TARGET, "Started {flavour} kvs store");
					Ok(TransactionBuilderParts::without_router_state(Box::<DatastoreFlavor>::new(
						v,
					)))
				}
				#[cfg(not(feature = "kv-surrealkv"))]
				bail!(Error::Kvs(crate::kvs::Error::Datastore("Cannot connect to the `surrealkv` storage engine as it is not enabled in this build of SurrealDB".to_owned())));
			}
			// Initiate an IndxDB database
			(flavour @ "indxdb", path) => {
				#[cfg(feature = "kv-indxdb")]
				{
					let v =
						super::indxdb::Datastore::new(&path).await.map(DatastoreFlavor::IndxDB)?;
					info!(target: TARGET, "Started {flavour} kvs store");
					Ok(TransactionBuilderParts::without_router_state(Box::<DatastoreFlavor>::new(
						v,
					)))
				}
				#[cfg(not(feature = "kv-indxdb"))]
				bail!(Error::Kvs(crate::kvs::Error::Datastore("Cannot connect to the `indxdb` storage engine as it is not enabled in this build of SurrealDB".to_owned())));
			}
			// Initiate a TiKV datastore
			(flavour @ "tikv", path) => {
				#[cfg(feature = "kv-tikv")]
				{
					// Parse TiKV-specific configuration from env vars
					// (SURREAL_TIKV_*) and query parameters.
					let tikv_config = config.load();
					let v = super::tikv::Datastore::new(&path, tikv_config)
						.await
						.map(DatastoreFlavor::TiKV)?;
					info!(target: TARGET, "Started {flavour} kvs store");
					Ok(TransactionBuilderParts::without_router_state(Box::<DatastoreFlavor>::new(
						v,
					)))
				}
				#[cfg(not(feature = "kv-tikv"))]
				bail!(Error::Kvs(crate::kvs::Error::Datastore("Cannot connect to the `tikv` storage engine as it is not enabled in this build of SurrealDB".to_owned())));
			}
			// The datastore path is not valid
			(flavour, path) => {
				info!(target: TARGET, "Unable to load the specified datastore {flavour}{path}");
				bail!(Error::Kvs(crate::kvs::Error::Datastore(
					"Unable to load the specified datastore".into()
				)))
			}
		}
	}

	fn path_valid(v: &str) -> Result<String> {
		// Strip query parameters before validating the scheme
		let scheme_part = v.split_once('?').map(|(s, _)| s).unwrap_or(v);
		match scheme_part {
			"memory" => Ok(v.to_string()),
			"mem" => Ok(v.to_string()),
			v_s if v_s.starts_with("file:") => Ok(v.to_string()),
			v_s if v_s.starts_with("rocksdb:") => Ok(v.to_string()),
			v_s if v_s.starts_with("surrealkv:") => Ok(v.to_string()),
			v_s if v_s.starts_with("mem:") => Ok(v.to_string()),
			v_s if v_s.starts_with("tikv:") => Ok(v.to_string()),
			_ => bail!("Provide a valid database path parameter"),
		}
	}
}

impl TransactionBuilderRequirements for DatastoreFlavor {}

impl TransactionBuilder for DatastoreFlavor {
	#[allow(
		unreachable_code,
		unreachable_patterns,
		unused_variables,
		reason = "Some variables are unused when no backends are enabled."
	)]
	fn new_transaction(
		&self,
		write: bool,
		lock: bool,
	) -> BoxFut<'_, Result<(Box<dyn Transactable>, bool)>> {
		Box::pin(async move {
			Ok(match self {
				#[cfg(feature = "kv-mem")]
				Self::Mem(v) => {
					let tx = v.transaction(write, lock).await?;
					(tx, true)
				}
				#[cfg(feature = "kv-rocksdb")]
				Self::RocksDB(v) => {
					let tx = v.transaction(write, lock).await?;
					(tx, true)
				}
				#[cfg(feature = "kv-indxdb")]
				Self::IndxDB(v) => {
					let tx = v.transaction(write, lock).await?;
					(tx, true)
				}
				#[cfg(feature = "kv-tikv")]
				Self::TiKV(v) => {
					let tx = v.transaction(write, lock).await?;
					(tx, false)
				}
				#[cfg(feature = "kv-surrealkv")]
				Self::SurrealKV(v) => {
					let tx = v.transaction(write, lock).await?;
					(tx, true)
				}
				_ => unreachable!(),
			})
		})
	}

	/// Registers metrics for the current datastore flavor if supported.
	fn register_metrics(&self) -> Option<Metrics> {
		match self {
			#[cfg(feature = "kv-rocksdb")]
			DatastoreFlavor::RocksDB(v) => Some(v.register_metrics()),
			#[allow(unreachable_patterns)]
			_ => None,
		}
	}

	/// Collects a specific u64 metric by name if supported by the datastore flavor.
	// Allow unused variable when kv-rocksdb feature is not enabled
	#[allow(unused_variables)]
	fn collect_u64_metric(&self, metric: &str) -> Option<u64> {
		match self {
			#[cfg(feature = "kv-rocksdb")]
			DatastoreFlavor::RocksDB(v) => v.collect_u64_metric(metric),
			#[allow(unreachable_patterns)]
			_ => None,
		}
	}

	fn shutdown(&self) -> BoxFut<'_, Result<()>> {
		Box::pin(async move {
			match self {
				#[cfg(feature = "kv-mem")]
				Self::Mem(v) => Ok(v.shutdown().await?),
				#[cfg(feature = "kv-rocksdb")]
				Self::RocksDB(v) => Ok(v.shutdown().await?),
				#[cfg(feature = "kv-indxdb")]
				Self::IndxDB(v) => Ok(v.shutdown().await?),
				#[cfg(feature = "kv-tikv")]
				Self::TiKV(v) => Ok(v.shutdown().await?),
				#[cfg(feature = "kv-surrealkv")]
				Self::SurrealKV(v) => Ok(v.shutdown().await?),
				#[allow(unreachable_patterns)]
				_ => unreachable!(),
			}
		})
	}

	#[allow(
		unused_variables,
		reason = "type_id is only consumed when a backend feature is enabled"
	)]
	fn extension(&self, type_id: TypeId) -> Option<Arc<dyn Any + Send + Sync>> {
		match self {
			#[cfg(feature = "kv-tikv")]
			Self::TiKV(v) if type_id == TypeId::of::<super::tikv::TikvOpsHandle>() => Some(v.ops_handle()),
			#[allow(unreachable_patterns)]
			_ => None,
		}
	}
}

impl Display for DatastoreFlavor {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		#![allow(unused_variables)]
		match self {
			#[cfg(feature = "kv-mem")]
			Self::Mem(_) => write!(f, "memory"),
			#[cfg(feature = "kv-rocksdb")]
			Self::RocksDB(_) => write!(f, "rocksdb"),
			#[cfg(feature = "kv-indxdb")]
			Self::IndxDB(_) => write!(f, "indxdb"),
			#[cfg(feature = "kv-tikv")]
			Self::TiKV(_) => write!(f, "tikv"),
			#[cfg(feature = "kv-surrealkv")]
			Self::SurrealKV(_) => write!(f, "surrealkv"),
			#[allow(unreachable_patterns)]
			_ => unreachable!(),
		}
	}
}

impl Display for Datastore {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		self.transaction_factory.builder.fmt(f)
	}
}

impl Datastore {
	pub fn builder() -> Builder {
		Builder::new()
	}

	async fn retry_index_operation_conflict(
		err: &anyhow::Error,
		operation: impl Into<String>,
	) -> bool {
		if is_retryable_transaction_conflict(err) {
			let operation = operation.into();
			debug!(
				target: TARGET,
				operation = %operation,
				error = %err,
				"retryable index operation conflict, retrying"
			);
			sleep(Duration::from_millis(100)).await;
			true
		} else {
			false
		}
	}

	async fn cancel_and_retry_index_operation_conflict(
		txn: &Transaction,
		err: &anyhow::Error,
		operation: impl Into<String>,
	) -> bool {
		let _ = txn.cancel().await;
		Self::retry_index_operation_conflict(err, operation).await
	}

	/// Creates a new datastore instance
	///
	/// # Examples
	///
	/// ```rust,no_run
	/// # use surrealdb_core::kvs::Datastore;
	/// # use anyhow::Error;
	/// # #[tokio::main]
	/// # async fn main() -> Result<(),Error> {
	/// let ds = Datastore::new("memory").await?;
	/// # Ok(())
	/// # }
	/// ```
	///
	/// Or to create a file-backed store:
	///
	/// ```rust,no_run
	/// # use surrealdb_core::kvs::Datastore;
	/// # use anyhow::Error;
	/// # #[tokio::main]
	/// # async fn main() -> Result<(),Error> {
	/// let ds = Datastore::new("surrealkv://temp.skv").await?;
	/// # Ok(())
	/// # }
	/// ```
	///
	/// Or to connect to a tikv-backed distributed store:
	///
	/// ```rust,no_run
	/// # use surrealdb_core::kvs::Datastore;
	/// # use anyhow::Error;
	/// # #[tokio::main]
	/// # async fn main() -> Result<(),Error> {
	/// let ds = Datastore::new("tikv://127.0.0.1:2379").await?;
	/// # Ok(())
	/// # }
	/// ```
	pub async fn new(path: &str) -> Result<Self> {
		Builder::new().build_with_path(path).await
	}

	/// Registers metrics for the current datastore flavor if supported.
	///
	/// This will return a list of available metrics and their descriptions.
	pub fn register_metrics(&self) -> Option<Metrics> {
		self.transaction_factory.register_metrics()
	}

	/// Collects a specific u64 metric by name if supported by the datastore flavor.
	///
	/// - `metric`: The name of the metric to collect.
	pub fn collect_u64_metric(&self, metric: &str) -> Option<u64> {
		self.transaction_factory.collect_u64_metric(metric)
	}

	/// The currently installed observer. Cheap to clone; internal handle is
	/// an `Arc`.
	///
	/// Exposed so higher layers (server, SDK) can emit transport-layer events
	/// such as session connect/disconnect that the core engine never sees.
	pub fn observer(&self) -> &Arc<dyn ExecutionObserver> {
		&self.observer
	}

	/// Create a new datastore with the same persistent data (inner), with
	/// flushed cache. Simulating a server restart
	pub fn restart(self) -> Self {
		self.buckets.clear();
		Self {
			id: self.id,
			auth_enabled: self.auth_enabled,
			dynamic_configuration: DynamicConfiguration::default(),
			slow_log: self.slow_log,
			transaction_timeout: self.transaction_timeout,
			capabilities: Arc::clone(&self.capabilities),
			live_query_broker: self.live_query_broker,
			// Fresh router cursor: a restarted node re-establishes its baseline.
			live_query_router: Arc::new(LiveQueryRouter::new()),
			http_endpoint: self.http_endpoint,
			index_stores: IndexStores::new(
				self.config.hnsw_cache_size,
				self.config.diskann_cache_size,
			),
			index_builder: IndexBuilder::new(self.transaction_factory.clone()),
			#[cfg(storage)]
			temporary_directory: self.temporary_directory,
			cache: Arc::new(DatastoreCache::new(self.config.datastore_cache_size)),
			#[cfg(all(feature = "graphql", not(target_family = "wasm")))]
			graphql_schema_cache: crate::graphql::cache::GraphQLSchemaCache::default(),
			function_registry: Arc::new(FunctionRegistry::with_builtins()),
			buckets: self.buckets,
			sequences: Sequences::new(self.transaction_factory.clone(), self.id),
			transaction_factory: self.transaction_factory,
			async_event_trigger: self.async_event_trigger,
			#[cfg(feature = "surrealism")]
			surrealism_cache: Arc::new(SurrealismCache::new(self.config.surrealism_cache_size)),
			#[cfg(feature = "surrealism")]
			lazy_surrealism: self.lazy_surrealism,
			#[cfg(feature = "http")]
			http_client: self.http_client,
			observer: self.observer,
			config: self.config,
		}
	}

	/// Create a test-only datastore facade that shares the same durable KV engine
	/// while resetting process-local state.
	///
	/// This lets unit tests model two SurrealDB compute nodes connected to the
	/// same storage backend without starting an external service. The cloned
	/// facade deliberately reuses the transaction factory, but gets its own node
	/// id, index builder, index stores, datastore cache, sequences, and other
	/// process-local caches. Tests that exercise cluster liveness should call
	/// [`Self::insert_node`] for both the original datastore and the fork.
	#[cfg(test)]
	#[cfg_attr(not(feature = "kv-mem"), allow(dead_code))]
	pub(crate) fn fork_for_test_with_node_id(&self, id: Uuid) -> Self {
		let transaction_factory = self.transaction_factory.clone();
		Self {
			id,
			auth_enabled: self.auth_enabled,
			dynamic_configuration: self.dynamic_configuration.clone(),
			slow_log: self.slow_log.clone(),
			transaction_timeout: self.transaction_timeout,
			capabilities: Arc::clone(&self.capabilities),
			live_query_broker: self.live_query_broker.clone(),
			// A fork models a separate node, so it gets its own router cursor.
			live_query_router: Arc::new(LiveQueryRouter::new()),
			http_endpoint: self.http_endpoint.clone(),
			index_stores: IndexStores::new(
				self.config.hnsw_cache_size,
				self.config.diskann_cache_size,
			),
			index_builder: IndexBuilder::new(transaction_factory.clone()),
			#[cfg(storage)]
			temporary_directory: self.temporary_directory.clone(),
			cache: Arc::new(DatastoreCache::new(self.config.datastore_cache_size)),
			#[cfg(all(feature = "graphql", not(target_family = "wasm")))]
			graphql_schema_cache: crate::graphql::cache::GraphQLSchemaCache::default(),
			function_registry: Arc::new(FunctionRegistry::with_builtins()),
			buckets: self.buckets.clone(),
			sequences: Sequences::new(transaction_factory.clone(), id),
			transaction_factory,
			async_event_trigger: Arc::clone(&self.async_event_trigger),
			#[cfg(feature = "surrealism")]
			surrealism_cache: Arc::new(SurrealismCache::new(self.config.surrealism_cache_size)),
			#[cfg(feature = "surrealism")]
			lazy_surrealism: self.lazy_surrealism,
			#[cfg(feature = "http")]
			http_client: Arc::clone(&self.http_client),
			observer: Arc::clone(&self.observer),
			config: Arc::clone(&self.config),
		}
	}

	/// Set the node id for this datastore.
	pub fn with_node_id(mut self, id: Uuid) -> Self {
		self.id = id;
		self
	}

	/// Set a global transaction timeout for this Datastore
	pub fn with_transaction_timeout(mut self, duration: Option<Duration>) -> Self {
		self.transaction_timeout = duration;
		self
	}

	/// Get the configured transaction timeout, if any
	pub(crate) fn transaction_timeout(&self) -> Option<Duration> {
		self.transaction_timeout
	}

	/// Returns the broker used to flush live-query notifications after commit.
	pub(crate) fn live_query_broker(&self) -> Option<Arc<dyn MessageBroker>> {
		self.live_query_broker.clone()
	}

	/// Looks up the public HTTP endpoint that the cluster member with `node_id`
	/// has published on its `Node` catalog row.
	///
	/// Returns `Ok(None)` when the row exists but no endpoint is set, or when
	/// no such node has registered. Used by clustered live-query relays to
	/// resolve the target node's address at delivery time without consulting
	/// any in-memory cluster topology.
	pub async fn lookup_node_endpoint(&self, node_id: Uuid) -> Result<Option<String>> {
		let txn = self.transaction(Read, Optimistic).await?;
		let key = crate::key::root::nd::Nd::new(node_id);
		let res = txn.get(&key, None).await?;
		// Always cancel a read transaction; we don't write through it.
		let _ = txn.cancel().await;
		Ok(res.and_then(|node: Node| node.http_endpoint))
	}

	#[cfg(storage)]
	/// Set a temporary directory for ordering of large result sets
	pub fn with_temporary_directory(mut self, path: Option<PathBuf>) -> Self {
		self.temporary_directory = path.map(Arc::new);
		self
	}

	/// Configure whether surrealism modules are loaded lazily on first use
	/// rather than eagerly at startup.
	#[cfg(feature = "surrealism")]
	pub fn with_lazy_surrealism(mut self, lazy: bool) -> Self {
		self.lazy_surrealism = lazy;
		self
	}

	/// Returns `true` if surrealism modules are loaded lazily.
	#[cfg(feature = "surrealism")]
	pub fn is_lazy_surrealism(&self) -> bool {
		self.lazy_surrealism
	}

	pub fn index_store(&self) -> &IndexStores {
		&self.index_stores
	}

	/// Is authentication enabled for this Datastore?
	pub fn is_auth_enabled(&self) -> bool {
		self.auth_enabled
	}

	pub fn id(&self) -> Uuid {
		self.id
	}

	/// Does the datastore allow excecuting an RPC method?
	pub(crate) fn allows_rpc_method(&self, method_target: &MethodTarget) -> bool {
		self.capabilities.allows_rpc_method(method_target)
	}

	/// Does the datastore allow requesting an HTTP route?
	/// This function needs to be public to allow access from the CLI crate.
	pub fn allows_http_route(&self, route_target: &RouteTarget) -> bool {
		self.capabilities.allows_http_route(route_target)
	}

	/// Is the user allowed to query?
	pub fn allows_query_by_subject(&self, subject: impl Into<ArbitraryQueryTarget>) -> bool {
		self.capabilities.allows_query(&subject.into())
	}

	/// Is the user allowed to invoke the `eval::*` functions?
	pub fn allows_eval_query_by_subject(&self, subject: impl Into<EvalQueryTarget>) -> bool {
		self.capabilities.allows_eval_query(&subject.into())
	}

	/// Does the datastore allow connections to a network target?
	#[cfg(feature = "jwks")]
	pub(crate) fn allows_network_target(&self, net_target: &NetTarget) -> bool {
		self.capabilities.allows_network_target(net_target)
	}

	/// Set specific capabilities for this Datastore
	pub fn get_capabilities(&self) -> &Capabilities {
		&self.capabilities
	}

	#[cfg(feature = "jwks")]
	pub(crate) fn cache(&self) -> &Arc<DatastoreCache> {
		&self.cache
	}

	pub(super) fn clock_now(&self) -> Timestamp {
		SystemClock::new().now()
	}

	// Initialise the cluster and run bootstrap utilities
	// Returns the current version and a flag indicating if this is a new datastore
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn check_version(&self) -> Result<(MajorVersion, bool)> {
		// Retry because concurrent instances may conflict when writing the version key
		let (version, is_new) = Self::retry("Check version", || self.get_version()).await?;
		// Check we are running the latest version
		if !version.is_latest() {
			bail!(Error::OutdatedStorageVersion {
				expected: MajorVersion::latest().into(),
				actual: version.into(),
			});
		}
		// Everything ok
		Ok((version, is_new))
	}

	// Initialise the cluster and run bootstrap utilities
	// Returns the current version and a flag indicating if this is a new datastore
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn get_version(&self) -> Result<(MajorVersion, bool)> {
		// Start a new writeable transaction
		let txn = self.transaction(Write, Optimistic).await?.enclose();
		// Create the key where the version is stored
		let key = crate::key::version::new();
		// Check if a version is already set in storage
		let val = match catch!(txn, txn.get(&key, None).await) {
			// There is a version set in the storage
			Some(val) => {
				// We didn't write anything, so just rollback
				catch!(txn, txn.cancel().await);
				// Return the current version
				(val, false)
			}
			// There is no version set in the storage
			None => {
				// Fetch any keys immediately following the version key
				let rng = crate::key::version::proceeding();
				let keys = catch!(txn, txn.keys(rng, 1, 0, None).await);
				// Check the storage if there are any other keys set
				let version = if keys.is_empty() {
					// There are no keys set in storage, so this is a new database
					MajorVersion::latest()
				} else {
					// There were keys in storage, so this is an upgrade.
					// Log the first key found for diagnostic purposes.
					warn!(
						target: TARGET,
						first_key = ?keys.first().map(|k| format!("{:?}", k)),
						"No version key found but existing data detected in storage. \
						 This storage contains data from a previous SurrealDB version. \
						 The server will not start until the data is migrated or removed."
					);
					MajorVersion::v1()
				};
				// Attempt to set the current version in storage
				catch!(txn, txn.replace(&key, &version).await);
				// We set the version, so commit the transaction
				catch!(txn, txn.commit().await);
				// Return the current version
				(version, true)
			}
		};
		// Everything ok
		Ok(val)
	}

	// --------------------------------------------------
	// Initialisation functions
	// --------------------------------------------------

	/// Setup the initial cluster access credentials
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn initialise_credentials(&self, user: &str, pass: &str) -> Result<()> {
		// Retry because concurrent instances may conflict when creating the root user
		Self::retry("Initialise credentials", || self.initialise_credentials_attempt(user, pass))
			.await
	}

	/// Single attempt to create the root user if none exists.
	/// Separated from `initialise_credentials` so it can be wrapped in the retry loop.
	async fn initialise_credentials_attempt(&self, user: &str, pass: &str) -> Result<()> {
		// Start a new writeable transaction
		let txn = self.transaction(Write, Optimistic).await?.enclose();
		// Fetch the root users from the storage
		let users = catch!(txn, txn.all_root_users(None).await);
		// Process credentials, depending on existing users
		if users.is_empty() {
			// Display information in the logs
			info!(target: TARGET, "Credentials were provided, and no root users were found. The root user '{user}' will be created");
			// Create and new root user definition
			let stm = DefineUserStatement::new_with_password(
				Base::Root,
				user.to_owned(),
				pass,
				INITIAL_USER_ROLE.to_owned(),
			);
			let opt = Options::new(&CommonConfig::default())
				.with_auth(Arc::new(Auth::for_root(Role::Owner)));
			let mut ctx = self.setup_ctx()?;
			ctx.set_transaction(Arc::clone(&txn));
			let ctx = ctx.freeze();
			let mut stack = TreeStack::new();
			let res = stack.enter(|stk| stm.compute(stk, &ctx, &opt, None)).finish().await;
			catch!(txn, res);
			// We added a user, so commit the transaction
			txn.commit().await
		} else {
			// Display information in the logs
			warn!(target: TARGET, "Credentials were provided, but existing root users were found. The root user '{user}' will not be created");
			warn!(target: TARGET, "Consider removing the --user and --pass arguments from the server start command");
			// We didn't write anything, so just rollback
			txn.cancel().await
		}
	}

	/// Setup the default namespace and database
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn initialise_defaults(&self, namespace: &str, database: &str) -> Result<()> {
		info!(target: TARGET, "This is a new SurrealDB instance. Initialising default namespace '{namespace}' and database '{database}'");
		// Create the SQL statement
		let sql = r"
			DEFINE NAMESPACE $namespace COMMENT 'Default namespace generated by SurrealDB';
			USE NS $namespace;
			DEFINE DATABASE $database COMMENT 'Default database generated by SurrealDB';
			DEFINE CONFIG DEFAULT NAMESPACE $namespace DATABASE $database;
		"
		.to_string();

		// Create the variables
		let vars = map! {
			"namespace".to_string() => namespace.to_string().into_value(),
			"database".to_string() => database.to_string().into_value(),
		};

		// Execute the SQL statement
		self.execute(
			&sql,
			&Session::owner(),
			Some(vars.into_iter().collect::<std::collections::BTreeMap<_, _>>().into()),
		)
		.await?;
		// Everything ok
		Ok(())
	}

	/// Performs a database import from SQL
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn startup(&self, sql: &str, sess: &Session) -> Result<Vec<QueryResult>> {
		// Output function invocation details to logs
		trace!(target: TARGET, "Running datastore startup import script");
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);
		// Execute the SQL import
		self.execute(sql, sess, None).await.map_err(|e| anyhow::anyhow!(e))
	}

	/// Run the datastore shutdown tasks, performing any necessary cleanup
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn shutdown(&self) -> Result<()> {
		// Output function invocation details to logs
		trace!(target: TARGET, "Running datastore shutdown operations");
		// Local index builder tasks are deliberately left running: the storage
		// engine shutdown below stops the commit coordinator first, after which
		// every commit is refused before it applies (see the engine
		// `Transaction::commit` gate). An in-flight build therefore makes no
		// durable change during shutdown and stays at its last committed
		// checkpoint, so the periodic resume scan continues it after restart —
		// exactly as it would after a crash.
		// Archive this datastore in the cluster, but don't let a blocked
		// metadata transaction prevent storage engine shutdown.
		let _ = archive_node_for_shutdown(
			NODE_DELETE_TIMEOUT,
			self.delete_node_with_timeout(NODE_DELETE_TIMEOUT).await,
		);
		// Run any storage engine shutdown tasks
		self.transaction_factory.builder.shutdown().await
	}

	/// Drop every version of every key in the half-open range `[start, end)`
	/// **outside** any user transaction.
	///
	/// Routes through the backend's [`TransactionBuilder::extension`] hook.
	/// Backends other than TiKV treat this as a no-op. Callers are
	/// responsible for ensuring the data is already logically inaccessible
	/// (typically by running this only after a committed catalog-clearing
	/// transaction).
	pub async fn unsafe_destroy_range(&self, start: Vec<u8>, end: Vec<u8>) -> Result<()> {
		#[cfg(feature = "kv-tikv")]
		if let Some(ops) = self.tikv_ops() {
			return ops.unsafe_destroy_range(start, end).await.map_err(Into::into);
		}
		let _ = (start, end);
		Ok(())
	}

	/// Advance the MVCC garbage-collection safepoint by `lifetime`.
	///
	/// Routes through the backend's [`TransactionBuilder::extension`] hook;
	/// no-op on backends other than TiKV. Background tasks call this on
	/// `EngineOptions::tikv_gc_interval` and shutdown runs one final
	/// advisory pass.
	pub async fn run_mvcc_gc(&self, lifetime: Duration) -> Result<()> {
		#[cfg(feature = "kv-tikv")]
		if let Some(ops) = self.tikv_ops() {
			return ops.run_mvcc_gc(lifetime).await.map_err(Into::into);
		}
		let _ = lifetime;
		Ok(())
	}

	/// Resolve stale transactional locks left by crashed clients.
	///
	/// Routes through the backend's [`TransactionBuilder::extension`] hook;
	/// no-op on backends other than TiKV. Background tasks call this on
	/// `EngineOptions::tikv_lock_cleanup_interval`.
	pub async fn run_lock_cleanup(&self, lifetime: Duration) -> Result<()> {
		#[cfg(feature = "kv-tikv")]
		if let Some(ops) = self.tikv_ops() {
			return ops.run_lock_cleanup(lifetime).await.map_err(Into::into);
		}
		let _ = lifetime;
		Ok(())
	}

	/// Number of in-flight transactions tracked by the backend, when
	/// available. `None` indicates the backend does not track this.
	pub fn in_flight_transaction_count(&self) -> Option<usize> {
		#[cfg(feature = "kv-tikv")]
		if let Some(ops) = self.tikv_ops() {
			return Some(ops.in_flight_transaction_count());
		}
		None
	}

	/// Resolve the TiKV operational extension handle, if the backend is
	/// TiKV. Returns `None` for every other flavour.
	#[cfg(feature = "kv-tikv")]
	fn tikv_ops(&self) -> Option<Arc<super::tikv::TikvOpsHandle>> {
		let ext = self
			.transaction_factory
			.builder
			.extension(TypeId::of::<super::tikv::TikvOpsHandle>())?;
		ext.downcast::<super::tikv::TikvOpsHandle>().ok()
	}

	// --------------------------------------------------
	// Surrealism eager loading
	// --------------------------------------------------

	/// Pre-load all Surrealism module runtimes into the cache so that
	/// subsequent query planning can resolve function metadata (e.g. the
	/// `writeable` flag) without triggering on-demand compilation.
	///
	/// Modules are loaded in parallel using a `JoinSet`. Any individual
	/// failure is logged but does not abort the overall loading process.
	#[cfg(feature = "surrealism")]
	pub async fn eager_load_surrealism_modules(&self) {
		use crate::catalog::providers::{DatabaseProvider, NamespaceProvider};
		use crate::surrealism::cache::SurrealismCacheLookup;

		let txn = match self.transaction(Read, Optimistic).await {
			Ok(txn) => Arc::new(txn),
			Err(e) => {
				warn!(target: TARGET, error = %e, "Surrealism eager load: failed to open transaction");
				return;
			}
		};

		let mut ctx = match self.setup_ctx() {
			Ok(ctx) => ctx,
			Err(e) => {
				warn!(target: TARGET, error = %e, "Surrealism eager load: failed to set up context");
				return;
			}
		};
		ctx.set_transaction(Arc::clone(&txn));
		let ctx = ctx.freeze();

		let nss = match txn.all_ns(None).await {
			Ok(nss) => nss,
			Err(e) => {
				warn!(target: TARGET, error = %e, "Surrealism eager load: failed to list namespaces");
				return;
			}
		};

		// Collect all module lookups first, then load in parallel.
		struct ModuleLookup {
			ns_id: crate::catalog::NamespaceId,
			db_id: crate::catalog::DatabaseId,
			bucket: String,
			key: String,
			display_name: String,
		}

		let mut lookups = Vec::new();
		for ns in nss.iter() {
			let dbs = match txn.all_db(ns.namespace_id, None).await {
				Ok(dbs) => dbs,
				Err(e) => {
					warn!(
						target: TARGET,
						error = %e, ns = %ns.name,
						"Surrealism eager load: failed to list databases"
					);
					continue;
				}
			};
			for db in dbs.iter() {
				let modules = match txn.all_db_modules(ns.namespace_id, db.database_id, None).await
				{
					Ok(m) => m,
					Err(e) => {
						warn!(
							target: TARGET,
							error = %e, ns = %ns.name, db = %db.name,
							"Surrealism eager load: failed to list modules"
						);
						continue;
					}
				};
				for md in modules.iter() {
					if let crate::catalog::ModuleExecutable::Surrealism(s) = &md.executable {
						lookups.push(ModuleLookup {
							ns_id: ns.namespace_id,
							db_id: db.database_id,
							bucket: s.bucket.clone(),
							key: s.key.clone(),
							display_name: md
								.name
								.clone()
								.unwrap_or_else(|| "<unnamed>".to_string()),
						});
					}
				}
			}
		}

		if lookups.is_empty() {
			debug!(target: TARGET, "Surrealism eager load: no modules to load");
			return;
		}

		let total = lookups.len();
		debug!(target: TARGET, count = total, "Surrealism eager load: loading modules");

		let concurrency =
			std::thread::available_parallelism().map(|n| n.get()).unwrap_or(8).clamp(2, 16);
		let load_sem = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency));

		let mut join_set = tokio::task::JoinSet::new();
		for lookup in lookups {
			let ctx = Arc::clone(&ctx);
			let load_sem = Arc::clone(&load_sem);
			join_set.spawn(async move {
				let _permit = load_sem
					.acquire_owned()
					.await
					.expect("Surrealism eager load semaphore must not be closed");
				let cache_lookup = SurrealismCacheLookup::File(
					&lookup.ns_id,
					&lookup.db_id,
					&lookup.bucket,
					&lookup.key,
				);
				match ctx.get_surrealism_runtime(cache_lookup).await {
					Ok(_) => {
						debug!(
							target: TARGET,
							module = %lookup.display_name,
							"Surrealism eager load: loaded module"
						);
						true
					}
					Err(e) => {
						warn!(
							target: TARGET,
							module = %lookup.display_name,
							error = %e,
							"Surrealism eager load: failed to load module"
						);
						false
					}
				}
			});
		}

		let mut loaded = 0usize;
		let mut failed = 0usize;
		while let Some(result) = join_set.join_next().await {
			match result {
				Ok(true) => loaded += 1,
				Ok(false) => failed += 1,
				Err(e) => {
					warn!(target: TARGET, error = %e, "Surrealism eager load: task panicked");
					failed += 1;
				}
			}
		}

		if failed > 0 {
			warn!(
				target: TARGET,
				loaded, failed, total,
				"Surrealism eager load: completed with failures"
			);
		} else {
			tracing::info!(
				target: TARGET,
				loaded, total,
				"Surrealism eager load: all modules loaded"
			);
		}
	}

	// --------------------------------------------------
	// Node functions
	// --------------------------------------------------

	/// Initialise the cluster and run bootstrap utilities
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn bootstrap(&self) -> Result<()> {
		// Each bootstrap step is retried independently, because concurrent instances
		// writing to the same cluster metadata keys may cause transaction conflicts.
		// Insert this node in the cluster
		Self::retry("Insert node", || self.insert_node()).await?;
		// Mark inactive nodes as archived
		Self::retry("Expire nodes", || self.expire_nodes()).await?;
		// Remove archived nodes
		Self::retry("Remove nodes", || self.remove_nodes()).await?;
		// Everything ok
		Ok(())
	}

	/// Retries an async operation until it succeeds or the global timeout elapses.
	///
	/// Only [`TransactionConflict`](crate::kvs::Error::TransactionConflict)
	/// errors are retried; any other error is returned immediately to the
	/// caller. On each retryable failure a randomized delay (0–10 s) is
	/// applied before the next attempt, adding jitter to reduce repeated
	/// collisions when multiple instances start concurrently against the
	/// same storage backend.
	///
	/// The global timeout is checked only after a *failed* attempt; a successful
	/// result is always returned immediately, even if the elapsed time
	/// exceeds the budget. Each attempt's timeout is the lesser of its
	/// natural timeout (10 s * attempt number) and the remaining global
	/// budget, so total wall-clock time never significantly exceeds the
	/// global timeout. If no attempt succeeds within the budget, an error
	/// is returned.
	async fn retry<F, Fut, R>(task: &str, func: F) -> Result<R>
	where
		F: Fn() -> Fut,
		Fut: Future<Output = Result<R>>,
	{
		let global_timeout = Duration::from_secs(120);
		let per_attempt_timeout = Duration::from_secs(10);
		let time = Instant::now();
		let mut last_error = None;
		let mut attempt = 1;
		loop {
			// Cap each attempt to the remaining global budget
			let remaining = global_timeout.saturating_sub(time.elapsed());
			if remaining.is_zero() {
				break;
			}
			let attempt_timeout = (per_attempt_timeout * attempt).min(remaining);
			if let Ok(result) = timeout(attempt_timeout, func()).await {
				match result {
					Ok(result) => return Ok(result),
					Err(e) => {
						// Only retry on transaction conflict errors
						if let Some(crate::kvs::Error::TransactionConflict(_)) = e.downcast_ref() {
							last_error = Some(e);
						} else {
							return Err(e);
						}
					}
				}
			}
			// Check if the global timeout has been exceeded
			if time.elapsed() >= global_timeout {
				break;
			}
			// Randomized back-off capped to the remaining budget
			let remaining = global_timeout.saturating_sub(time.elapsed());
			if remaining.is_zero() {
				break;
			}
			let tempo = Duration::from_secs(rand::rng().random_range(0..10)).min(remaining);
			sleep(tempo).await;
			attempt += 1;
		}
		if let Some(e) = last_error {
			error!(target: TARGET, "{task} - All {attempt} attempts failed. Last error: {e}");
		} else {
			error!(target: TARGET, "{task} - All {attempt} attempts failed.");
		}
		bail!(Error::Internal(format!("{task} failed after {attempt} attempts due to timeout")));
	}

	/// Registers this node's cluster membership entry with a fresh heartbeat.
	///
	/// Must be run at server or database startup. The write is idempotent:
	/// the entry at `Nd::new(self.id)` is owned by this node, so the call
	/// upserts the row whether or not a previous lifetime of the same node
	/// id left a record behind. This supports deployments where the node id
	/// is stable across restarts (e.g. a stateful cluster member reusing
	/// its durable storage) without forcing the operator to clean up state
	/// between runs.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn insert_node(&self) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, id = %self.id,"Inserting node in the cluster");
		// Refresh system usage metrics
		crate::sys::refresh().await;
		// Open transaction and set node data
		let txn = self.transaction(Write, Optimistic).await?;
		let key = crate::key::root::nd::Nd::new(self.id);
		let now = self.clock_now();
		let node = Node::new_with_endpoint(self.id, now, false, self.http_endpoint.clone());
		run!(txn, txn.set(&key, &node).await)
	}

	/// Updates an already existing node in the cluster.
	///
	/// This function should be run periodically at a regular interval.
	///
	/// This function updates the entry for this node with an up-to-date
	/// timestamp. This ensures that the node is not marked as expired by any
	/// garbage collection tasks, preventing any data cleanup for this node.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn update_node(&self) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, id = %self.id, "Updating node in the cluster");
		// Refresh system usage metrics
		crate::sys::refresh().await;
		// Open transaction and set node data
		let txn = self.transaction(Write, Optimistic).await?;
		let key = crate::key::root::nd::new(self.id);
		let now = self.clock_now();
		let node = Node::new_with_endpoint(self.id, now, false, self.http_endpoint.clone());
		run!(txn, txn.replace(&key, &node).await)
	}

	/// Updates this node, bounding each step and explicitly cancelling any
	/// open write transaction before returning on timeout or cancellation.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self, canceller))]
	pub async fn update_node_with_timeout(
		&self,
		timeout_duration: Duration,
		canceller: &CancellationToken,
	) -> Result<()> {
		trace!(target: TARGET, id = %self.id, timeout = ?timeout_duration, "Updating node in the cluster with timeout");

		let deadline = Instant::now() + timeout_duration;

		await_node_step(deadline, timeout_duration, Some(canceller), async {
			crate::sys::refresh().await;
			Ok(())
		})
		.await?;

		let txn = await_node_step(
			deadline,
			timeout_duration,
			Some(canceller),
			self.transaction(Write, Optimistic),
		)
		.await?;
		let key = crate::key::root::nd::new(self.id);
		let now = self.clock_now();
		let node = Node::new_with_endpoint(self.id, now, false, self.http_endpoint.clone());

		await_node_tx_step(
			&txn,
			deadline,
			timeout_duration,
			Some(canceller),
			txn.replace(&key, &node),
		)
		.await?;
		await_node_tx_step(&txn, deadline, timeout_duration, Some(canceller), txn.commit()).await
	}

	/// Deletes a node from the cluster.
	///
	/// This function should be run when a node is shutting down.
	///
	/// This function marks the node as archived, ready for garbage collection.
	/// Later on when garbage collection is running the live queries assigned
	/// to this node will be removed, along with the node itself.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn delete_node(&self) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, id = %self.id, "Archiving node in the cluster");
		// Open transaction and set node data
		let txn = self.transaction(Write, Optimistic).await?;
		let key = crate::key::root::nd::new(self.id);
		let val = catch!(txn, txn.get_node(self.id).await);
		let node = val.as_ref().archive();
		run!(txn, txn.replace(&key, &node).await)
	}

	/// Archives this node, bounding each step and explicitly cancelling any
	/// open write transaction before returning on timeout.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn delete_node_with_timeout(&self, timeout_duration: Duration) -> Result<()> {
		trace!(target: TARGET, id = %self.id, timeout = ?timeout_duration, "Archiving node in the cluster with timeout");

		let deadline = Instant::now() + timeout_duration;
		let txn =
			await_node_step(deadline, timeout_duration, None, self.transaction(Write, Optimistic))
				.await?;
		let key = crate::key::root::nd::new(self.id);
		let val = await_node_tx_step(&txn, deadline, timeout_duration, None, txn.get_node(self.id))
			.await?;
		let node = val.as_ref().archive();

		await_node_tx_step(&txn, deadline, timeout_duration, None, txn.replace(&key, &node))
			.await?;
		await_node_tx_step(&txn, deadline, timeout_duration, None, txn.commit()).await
	}

	/// Expires nodes which have timedout from the cluster.
	///
	/// This function should be run periodically at an interval.
	///
	/// This function marks the node as archived, ready for garbage collection.
	/// Later on when garbage collection is running the live queries assigned
	/// to this node will be removed, along with the node itself.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn expire_nodes(&self) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, "Archiving expired nodes in the cluster");
		// Fetch all of the inactive nodes
		let inactive = {
			let txn = self.transaction(Read, Optimistic).await?;
			let nds = catch!(txn, txn.all_nodes().await);
			let now = self.clock_now();
			catch!(txn, txn.cancel().await);
			// Filter the inactive nodes
			nds.iter()
				.filter_map(|n| {
					// Check that the node is active and has expired
					match n.is_active() && n.heartbeat < now - Duration::from_secs(30) {
						true => Some(n.to_owned()),
						false => None,
					}
				})
				.collect::<Vec<_>>()
		};
		// Check if there are inactive nodes
		if !inactive.is_empty() {
			// Open a writeable transaction
			let txn = self.transaction(Write, Optimistic).await?;
			// Archive the inactive nodes
			for nd in inactive.iter() {
				// Log the live query scanning
				trace!(target: TARGET, id = %nd.id, "Archiving node in the cluster");
				// Mark the node as archived
				let node = nd.archive();
				// Get the key for the node entry
				let key = crate::key::root::nd::new(nd.id);
				// Update the node entry
				catch!(txn, txn.replace(&key, &node).await);
			}
			// Commit the changes
			catch!(txn, txn.commit().await);
		}
		// Everything was successful
		Ok(())
	}

	/// Removes and cleans up nodes which are no longer in this cluster.
	///
	/// This function should be run periodically at an interval.
	///
	/// This function clears up all nodes which have been marked as archived.
	/// When a matching node is found, all node queries, and table queries are
	/// garbage collected, before the node itself is completely deleted.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn remove_nodes(&self) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, "Cleaning up archived nodes in the cluster");
		// Fetch all of the archived nodes
		let archived = {
			let txn = self.transaction(Read, Optimistic).await?;
			let nds = catch!(txn, txn.all_nodes().await);
			catch!(txn, txn.cancel().await);
			// Filter the archived nodes
			nds.iter().filter_map(Node::archived).collect::<Vec<_>>()
		};
		// Loop over the archived nodes
		for id in archived.iter() {
			// Open a writeable transaction
			let beg = crate::key::node::lq::prefix(*id)?;
			let end = crate::key::node::lq::suffix(*id)?;
			let mut next = Some(beg..end);
			let txn = self.transaction(Write, Optimistic).await?;
			{
				// Log the live query scanning
				trace!(target: TARGET, id = %id, "Deleting live queries for node");
				// Scan the live queries for this node
				while let Some(rng) = next {
					// Fetch the next batch of keys and values
					let res = catch!(txn, txn.batch_keys_vals(rng, NORMAL_BATCH_SIZE, None).await);
					next = res.next;
					for (k, v) in res.result.iter() {
						// Decode the data for this live query
						let val: NodeLiveQuery = KVValue::kv_decode_value(v, ())?;
						// Get the key for this node live query
						let nlq = catch!(txn, crate::key::node::lq::Lq::decode_key(k));
						// Check that the node for this query is archived
						if archived.contains(&nlq.nd) {
							// Get the key for this table live query
							let tlq = crate::key::table::lq::new(val.ns, val.db, &val.tb, nlq.lq);
							// Delete the table live query
							catch!(txn, txn.clr(&tlq).await);
							// Delete the node live query
							catch!(txn, txn.clr(&nlq).await);
						}
					}
					// Pause and yield execution
					yield_now!();
				}
			}
			{
				// Log the node deletion
				trace!(target: TARGET, id = %id, "Deleting node from the cluster");
				// Get the key for the node entry
				let key = crate::key::root::nd::new(*id);
				// Delete the cluster node entry
				catch!(txn, txn.clr(&key).await);
			}
			// Commit the changes
			catch!(txn, txn.commit().await);
		}
		// Everything was successful
		Ok(())
	}

	// ------------------------------
	// Durable RPC session functions
	// ------------------------------

	/// Creates or replaces the durable copy of a client-attached RPC session,
	/// expiring `ttl` from now (an unconditional upsert).
	///
	/// Session creation goes through [`create_rpc_session`](Self::create_rpc_session)
	/// (insert-if-absent), local changes through
	/// [`update_rpc_session`](Self::update_rpc_session) (update-if-present),
	/// and cross-node reconciliation through
	/// [`load_rpc_session`](Self::load_rpc_session), so none resurrects or
	/// duplicates a session across nodes. This blind upsert is retained as a
	/// primitive for callers that own the id exclusively (and for test setup).
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self, session))]
	pub async fn persist_rpc_session(
		&self,
		id: Uuid,
		session: &Session,
		ttl: Duration,
	) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, id = %id, "Persisting durable RPC session");
		// Capture the durable form with a refreshed expiry
		let expires_at = self.clock_now().value + ttl.as_millis() as u64;
		let value = DurableSession::from_session(session, expires_at);
		// Open transaction and set the session data
		let key = crate::key::root::se::Se::new(id);
		let txn = self.transaction(Write, Optimistic).await?;
		run!(txn, txn.set(&key, &value).await)
	}

	/// Creates the durable copy of a client-attached RPC session, but only if
	/// one does not already exist. Returns `true` if it created the entry,
	/// `false` if a session with this id already exists.
	///
	/// Uses a conditional insert (`put_compare_key` with no expected value),
	/// which is atomic even on last-writer-wins backends (TiKV) — a blind
	/// `set` is not — so two concurrent `attach`es for the same id on
	/// different nodes cannot both create it: one wins and the other is
	/// reported as an existing session, letting `attach` return
	/// `session_exists` instead of silently overwriting.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self, session))]
	pub async fn create_rpc_session(
		&self,
		id: Uuid,
		session: &Session,
		ttl: Duration,
	) -> Result<bool> {
		trace!(target: TARGET, id = %id, "Creating durable RPC session");
		let key = crate::key::root::se::Se::new(id);
		let expires_at = self.clock_now().value + ttl.as_millis() as u64;
		let value = DurableSession::from_session(session, expires_at);
		let txn = self.transaction(Write, Optimistic).await?;
		// `put_compare_key(.., None)` writes only if the key is absent.
		match txn.putc(&key, &value, None).await {
			Ok(()) => match txn.commit().await {
				Ok(()) => Ok(true),
				// Cancel after any failed commit (a conflict or otherwise), so
				// the transaction is rolled back consistently with the `run!`
				// macro; only the error classification differs.
				Err(e) if is_conditional_write_conflict(&e) => {
					let _ = txn.cancel().await;
					Ok(false)
				}
				Err(e) => {
					let _ = txn.cancel().await;
					Err(e)
				}
			},
			Err(e) => {
				let _ = txn.cancel().await;
				if is_conditional_write_conflict(&e) {
					Ok(false)
				} else {
					Err(e)
				}
			}
		}
	}

	/// Refreshes the durable copy of an *existing* RPC session with the given
	/// value, expiring `ttl` from now, **without recreating it** if it is gone.
	///
	/// Returns `true` if the entry was present and updated, `false` if it no
	/// longer exists (or was changed on another node). Uses a conditional
	/// write (`put_compare_key` guarded by the value just read), so it is
	/// atomic on last-writer-wins backends (TiKV): a session deleted elsewhere
	/// between the read and the write is neither resurrected nor clobbered — a
	/// blind `set` would be last-writer-wins and could resurrect it. This is
	/// the "push a local change to storage" path (a session-mutating method);
	/// the authoritative *reload* path is
	/// [`load_rpc_session`](Self::load_rpc_session).
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self, session))]
	pub async fn update_rpc_session(
		&self,
		id: Uuid,
		session: &Session,
		ttl: Duration,
	) -> Result<bool> {
		trace!(target: TARGET, id = %id, "Updating durable RPC session");
		let key = crate::key::root::se::Se::new(id);
		let txn = self.transaction(Write, Optimistic).await?;
		// Read the current stored value to use as the conditional-write guard.
		let Some(current) = catch!(txn, txn.get(&key, None).await) else {
			let _ = txn.cancel().await;
			return Ok(false);
		};
		let expires_at = self.clock_now().value + ttl.as_millis() as u64;
		let value = DurableSession::from_session(session, expires_at);
		// Write only if the stored value is still the one we read, so a delete
		// or change committed on another node in between wins (no resurrection,
		// no clobber) instead of being overwritten by this blind write.
		match txn.putc(&key, &value, Some(&current)).await {
			Ok(()) => match txn.commit().await {
				Ok(()) => Ok(true),
				// Cancel after any failed commit (a conflict or otherwise), so
				// the transaction is rolled back consistently with the `run!`
				// macro; only the error classification differs.
				Err(e) if is_conditional_write_conflict(&e) => {
					let _ = txn.cancel().await;
					Ok(false)
				}
				Err(e) => {
					let _ = txn.cancel().await;
					Err(e)
				}
			},
			Err(e) => {
				let _ = txn.cancel().await;
				if is_conditional_write_conflict(&e) {
					Ok(false)
				} else {
					Err(e)
				}
			}
		}
	}

	/// Loads the current durable copy of a client-attached RPC session — the
	/// authoritative cross-node state — returning the restored session and its
	/// absolute expiry (ms since the UNIX epoch), or `None` if it is gone or
	/// expired.
	///
	/// This is a pure read plus lazy expiry: an entry past its `expires_at` is
	/// deleted and reported as a miss, but a live entry is returned **without**
	/// being rewritten. It never extends a session's life, so it is safe to
	/// call before the caller is authorized — both when `get_session`
	/// rehydrates an evicted session and when the transport reconciles a
	/// cached session against storage before dispatch. The TTL is slid only
	/// after authorization, by [`update_rpc_session`](Self::update_rpc_session).
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn load_rpc_session(&self, id: Uuid) -> Result<Option<(Session, u64)>> {
		// Log when this method is run
		trace!(target: TARGET, id = %id, "Loading durable RPC session");
		let key = crate::key::root::se::Se::new(id);
		let now = self.clock_now().value;
		// Pure read.
		let durable = {
			let txn = self.transaction(Read, Optimistic).await?;
			let val = catch!(txn, txn.get(&key, None).await);
			catch!(txn, txn.cancel().await);
			val
		};
		let Some(durable) = durable else {
			return Ok(None);
		};
		// Lazily delete an expired entry with a conditional delete guarded by
		// the value we read, so a refresh committed on another node between the
		// read and the delete is not clobbered (a blind `clr` is
		// last-writer-wins on TiKV). On a condition conflict the entry changed
		// — leave it and report a miss for this request; it rehydrates fresh on
		// the next one.
		if durable.expires_at <= now {
			trace!(target: TARGET, id = %id, "Durable RPC session has expired");
			let txn = self.transaction(Write, Optimistic).await?;
			match txn.delc(&key, Some(&durable)).await {
				Ok(()) => match txn.commit().await {
					Ok(()) => {}
					Err(e) if is_conditional_write_conflict(&e) => {
						let _ = txn.cancel().await;
					}
					Err(e) => {
						let _ = txn.cancel().await;
						return Err(e);
					}
				},
				Err(e) => {
					let _ = txn.cancel().await;
					if !is_conditional_write_conflict(&e) {
						return Err(e);
					}
				}
			}
			return Ok(None);
		}
		// Restore the in-memory session without touching the durable copy,
		// handing back the stored expiry so the caller can enforce the TTL.
		let expires_at = durable.expires_at;
		durable.into_session().map(|session| Some((session, expires_at)))
	}

	/// Deletes the durable copy of a client-attached RPC session, when it is
	/// still the value being torn down.
	///
	/// Run on detach/teardown so the session cannot be rehydrated. It reads the
	/// current value and deletes only that (`del_compare_key`), rather than
	/// clearing the key blindly: a blind `clr` is last-writer-wins on TiKV, so
	/// a slow detach could otherwise delete a *newer* session that another node
	/// re-attached under the same id after the old one was removed. An entry
	/// already absent is a no-op.
	///
	/// A condition conflict — the stored value changed between the read and the
	/// delete (a concurrent refresh/re-attach on another node) — is surfaced as
	/// an **error**, not swallowed: the durable copy was *not* removed, so
	/// reporting success would let [`RpcProtocol::forget_session`]/`detach`
	/// claim the session was torn down while it remains rehydratable. The
	/// caller (whose `del_session` removes the durable copy before local state)
	/// then fails the teardown and can retry, keeping detach fail-closed. In
	/// the supported single-node-per-session model such a conflict does not
	/// arise (the owning node serializes the session's writes).
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn delete_rpc_session(&self, id: Uuid) -> Result<()> {
		// Log when this method is run
		trace!(target: TARGET, id = %id, "Deleting durable RPC session");
		let key = crate::key::root::se::Se::new(id);
		let txn = self.transaction(Write, Optimistic).await?;
		// Read the value we intend to remove; if it is already gone, done.
		let Some(current) = catch!(txn, txn.get(&key, None).await) else {
			let _ = txn.cancel().await;
			return Ok(());
		};
		// Delete only if the stored value is still the one we read. A conflict
		// (value changed concurrently) means the delete did not happen, so it
		// is surfaced as an error rather than reported as a successful teardown.
		match txn.delc(&key, Some(&current)).await {
			Ok(()) => match txn.commit().await {
				Ok(()) => Ok(()),
				Err(e) => {
					let _ = txn.cancel().await;
					Err(e)
				}
			},
			Err(e) => {
				let _ = txn.cancel().await;
				Err(e)
			}
		}
	}

	/// Purges expired durable RPC sessions.
	///
	/// This function should be run periodically at a regular interval.
	///
	/// It uses a distributed task lease so only one node in the cluster
	/// performs the purge per interval. Entries are scanned in batches, and
	/// only entries whose expiry has passed are deleted; an entry that fails
	/// to decode (for example one written by a newer node during a rolling
	/// upgrade) is skipped, never deleted.
	///
	/// # Arguments
	/// * `interval` - The interval between purge runs, to calculate the lease duration
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn purge_expired_rpc_sessions(&self, interval: &Duration) -> Result<()> {
		// Output function invocation details to logs
		trace!(target: TARGET, "Attempting expired RPC session purge");
		// Create a new lease handler
		let lh = LeaseHandler::new(
			self.sequences.clone(),
			self.id,
			self.transaction_factory.clone(),
			TaskLeaseType::RpcSessionCleanup,
			*interval * 2,
		)?;
		// If we don't get the lease, another node is handling this task
		if !lh.has_lease().await? {
			return Ok(());
		}
		// Output function invocation details to logs
		trace!(target: TARGET, "Purging expired RPC sessions");
		let now = self.clock_now().value;
		// Scan the durable session entries in batches under read transactions,
		// deleting each expired entry in its own conditional-delete write
		// transaction. The delete is guarded by the value read
		// (`del_compare_key`), so a request that refreshed the session after
		// the scan read the old expired value is not clobbered — a blind `clr`
		// is last-writer-wins on TiKV. A per-entry transaction keeps one such
		// conflict from aborting the whole purge.
		let mut next = Some(crate::key::root::se::SePrefix {}.encode_range()?);
		while let Some(rng) = next {
			// Fetch the next batch of keys and values under a read transaction.
			let batch = {
				let txn = self.transaction(Read, Optimistic).await?;
				let res = catch!(txn, txn.batch_keys_vals(rng, NORMAL_BATCH_SIZE, None).await);
				catch!(txn, txn.cancel().await);
				res
			};
			next = batch.next;
			for (k, v) in batch.result.iter() {
				// Decode the data for this session entry. Skip (but never
				// delete) an entry we cannot decode: it may have been written
				// by a newer node during a rolling upgrade.
				let val: DurableSession = match KVValue::kv_decode_value(v, ()) {
					Ok(val) => val,
					Err(e) => {
						warn!(target: TARGET, "Skipping undecodable durable RPC session entry: {e}");
						continue;
					}
				};
				// Only delete an expired entry, and only if it is still the
				// value we read.
				if val.expires_at <= now {
					let key = crate::key::root::se::Se::decode_key(k)?;
					trace!(target: TARGET, id = %key.id, "Purging expired RPC session");
					let txn = self.transaction(Write, Optimistic).await?;
					match txn.delc(&key, Some(&val)).await {
						Ok(()) => match txn.commit().await {
							Ok(()) => {}
							// Refreshed concurrently at commit (TiKV): leave it.
							Err(e) if is_conditional_write_conflict(&e) => {
								let _ = txn.cancel().await;
							}
							Err(e) => {
								let _ = txn.cancel().await;
								return Err(e);
							}
						},
						Err(e) => {
							let _ = txn.cancel().await;
							if !is_conditional_write_conflict(&e) {
								return Err(e);
							}
						}
					}
				}
			}
			// Pause and yield execution
			yield_now!();
		}
		// Everything was successful
		Ok(())
	}

	/// Clean up all other miscellaneous data.
	///
	/// This function should be run periodically at an interval.
	///
	/// This function clears up all data which might have been missed from
	/// previous cleanup runs, or when previous runs failed. This function
	/// currently deletes all live queries, for nodes which no longer exist
	/// in the cluster, from all namespaces, databases, and tables. It uses
	/// a number of transactions in order to prevent failure of large or
	/// long-running transactions on distributed storage engines.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn garbage_collect(&self) -> Result<()> {
		// Log the node deletion
		trace!(target: TARGET, "Garbage collecting all miscellaneous data");
		// Fetch archived nodes
		let archived = {
			let txn = self.transaction(Read, Optimistic).await?;
			let nds = catch!(txn, txn.all_nodes().await);
			txn.cancel().await?;
			// Filter the archived nodes
			nds.iter().filter_map(Node::archived).collect::<Vec<_>>()
		};
		// Fetch all namespaces
		let nss = {
			let txn = self.transaction(Read, Optimistic).await?;
			let res = catch!(txn, txn.all_ns(None).await);
			txn.cancel().await?;
			res
		};
		// Loop over all namespaces
		for ns in nss.iter() {
			// Log the namespace
			trace!(target: TARGET, "Garbage collecting data in namespace {}", ns.name);
			// Fetch all databases
			let dbs = {
				let txn = self.transaction(Read, Optimistic).await?;
				let res = catch!(txn, txn.all_db(ns.namespace_id, None).await);
				txn.cancel().await?;
				res
			};
			// Loop over all databases
			for db in dbs.iter() {
				// Log the namespace
				trace!(target: TARGET, "Garbage collecting data in database {}/{}", ns.name, db.name);
				// Fetch all tables
				let tbs = {
					let txn = self.transaction(Read, Optimistic).await?;
					let res = catch!(txn, txn.all_tb(ns.namespace_id, db.database_id, None).await);
					txn.cancel().await?;
					res
				};
				// Loop over all tables
				for tb in tbs.iter() {
					// Log the namespace
					trace!(target: TARGET, "Garbage collecting data in table {}/{}/{}", ns.name, db.name, tb.name);
					// Iterate over the table live queries
					let beg =
						crate::key::table::lq::prefix(db.namespace_id, db.database_id, &tb.name)?;
					let end =
						crate::key::table::lq::suffix(db.namespace_id, db.database_id, &tb.name)?;
					let mut next = Some(beg..end);
					let txn = self.transaction(Write, Optimistic).await?;
					while let Some(rng) = next {
						// Fetch the next batch of keys and values
						let max = NORMAL_BATCH_SIZE;
						let res = catch!(txn, txn.batch_keys_vals(rng, max, None).await);
						next = res.next;
						for (k, v) in res.result.iter() {
							// Decode the LIVE query statement
							let stm: SubscriptionDefinition = KVValue::kv_decode_value(v, ())?;
							// Get the node id and the live query id
							let (nid, lid) = (stm.node, stm.id);
							// Check that the node for this query is archived
							if archived.contains(&stm.node) {
								// Get the key for this node live query
								let tlq = catch!(txn, crate::key::table::lq::Lq::decode_key(k));
								// Get the key for this table live query
								let nlq = crate::key::node::lq::new(nid, lid);
								// Delete the node live query
								catch!(txn, txn.clr(&nlq).await);
								// Delete the table live query
								catch!(txn, txn.clr(&tlq).await);
							}
						}
						// Pause and yield execution
						yield_now!();
					}
					// Commit the changes
					catch!(txn, txn.commit().await);
				}
			}
		}
		// All ok
		Ok(())
	}

	// --------------------------------------------------
	// Live query functions
	// --------------------------------------------------

	/// Clean up the live queries for a disconnected connection.
	///
	/// This function should be run when a WebSocket disconnects.
	///
	/// This function clears up the live queries on the current node, which
	/// are specified by uique live query UUIDs. This is necessary when a
	/// WebSocket disconnects, and any associated live queries need to be
	/// cleaned up and removed.
	#[instrument(err, level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn delete_queries(&self, ids: Vec<uuid::Uuid>) -> Result<()> {
		// Log the node deletion
		trace!(target: TARGET, "Deleting live queries for a connection");
		// Fetch expired nodes
		let txn = self.transaction(Write, Optimistic).await?;
		// Loop over the live query unique ids
		for id in ids {
			// Get the key for this node live query
			let nlq = crate::key::node::lq::new(self.id(), id);
			// Fetch the LIVE meta data node entry
			if let Some(lq) = catch!(txn, txn.get(&nlq, None).await) {
				// Get the key for this node live query
				let nlq = crate::key::node::lq::new(self.id(), id);
				// Get the key for this table live query
				let tlq = crate::key::table::lq::new(lq.ns, lq.db, &lq.tb, id);
				// Delete the table live query
				catch!(txn, txn.clr(&tlq).await);
				// Delete the node live query
				catch!(txn, txn.clr(&nlq).await);
			}
		}
		// Commit the changes
		catch!(txn, txn.commit().await);
		// All ok
		Ok(())
	}

	// --------------------------------------------------
	// Changefeed functions
	// --------------------------------------------------

	/// Performs changefeed garbage collection as a background task.
	///
	/// This method is responsible for cleaning up old changefeed data across
	/// all databases. It uses a distributed task lease mechanism to coordinate
	/// which node performs this maintenance operation. Once a batch starts it
	/// runs to completion even if the lease expires, so brief overlap is
	/// possible.
	///
	/// The process involves:
	/// 1. Acquiring a lease for the ChangeFeedCleanup task
	/// 2. Cleaning up old changefeed data from all databases
	///
	/// # Arguments
	/// * `interval` - The interval between compaction runs, to calculate the lease duration
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn changefeed_process(&self, interval: &Duration) -> Result<()> {
		// Output function invocation details to logs
		trace!(target: TARGET, "Attempting changefeed garbage collection");
		// Create a new lease handler
		let lh = LeaseHandler::new(
			self.sequences.clone(),
			self.id,
			self.transaction_factory.clone(),
			TaskLeaseType::ChangeFeedCleanup,
			*interval * 2,
		)?;
		// If we don't get the lease, another node is handling this task
		if !lh.has_lease().await? {
			return Ok(());
		}
		// Output function invocation details to logs
		trace!(target: TARGET, "Running changefeed garbage collection");
		// Create a new transaction
		let txn = self.transaction(Write, Optimistic).await?;
		// Perform the changefeed garbage collection
		catch!(txn, crate::cf::gc_all_at(&lh, &txn).await);
		// When the Router engine is active, also garbage-collect the dedicated
		// live-query event keyspace, retaining entries for the configured window so
		// reconnecting/lagging subscribers can still replay. This rides the same
		// lease and transaction as the changefeed GC above.
		if self.config.live_query_engine == LiveQueryEngine::Router {
			catch!(
				txn,
				crate::lq::gc::gc_all_at(&lh, &txn, self.config.live_query_retention).await
			);
		}
		// Commit the changes
		catch!(txn, txn.commit().await);
		// Everything ok
		Ok(())
	}

	/// Run one live-query router pass.
	///
	/// Under the [`LiveQueryEngine::Router`] engine this tails the dedicated
	/// `lqe` keyspace since the node's cursor and delivers notifications off the
	/// write path (see [`crate::lq::router`]); it is the sole delivery path in
	/// that mode. Under the default [`LiveQueryEngine::Inline`] engine it is a
	/// cheap no-op, so the engine's background task can call it unconditionally.
	/// Unlike changefeed GC this runs on every node without a lease: each node
	/// delivers only to the subscriptions it owns.
	#[instrument(level = "trace", target = "surrealdb::core::lq", skip(self))]
	pub async fn live_query_router_process(&self) -> Result<()> {
		// Only the Router engine delivers via the router.
		if self.config.live_query_engine != LiveQueryEngine::Router {
			return Ok(());
		}
		crate::lq::router::process(self, &self.live_query_router).await
	}

	// --------------------------------------------------
	// Indexing functions
	// --------------------------------------------------

	fn ensure_not_cancelled(canceller: &CancellationToken) -> Result<()> {
		if canceller.is_cancelled() {
			bail!(Error::QueryCancelled);
		}
		Ok(())
	}

	/// Processes the index compaction queue.
	///
	/// This method is called periodically by the index compaction thread to
	/// process indexes that have been marked for compaction. It acquires a
	/// distributed lease to coordinate compaction across the cluster. Once a
	/// batch starts it runs to completion even if the lease expires, so brief
	/// overlap is possible.
	///
	/// The method scans the index compaction queue (stored as `Ic` keys) and
	/// delegates to [`Self::index_compaction_loop`], which compacts each
	/// distinct index exactly once — duplicate queue entries for the same
	/// index are skipped. On native targets compaction tasks run in parallel
	/// (one spawned task per index), while on wasm they run sequentially.
	/// Indexes that support compaction include full-text, count, and HNSW.
	///
	/// The queue is read in a short-lived read transaction so that user
	/// transactions enqueueing new compaction requests do not conflict with
	/// the compaction cycle. Each index compaction runs on its own write
	/// transaction. Once all compactions have completed, a separate write
	/// transaction removes the processed queue entries. Compaction failures
	/// are logged but do not prevent other indexes from being processed.
	///
	/// # Arguments
	/// * `dbs` - The shared datastore instance, cloned into each compaction task
	/// * `interval` - The interval between compaction runs, used to calculate the lease duration
	/// * `canceller` - Token checked before starting each lease, batch, and compaction unit
	///
	/// # Returns
	/// A tuple `(iterations, errors)` where `iterations` is the number of
	/// compaction batches processed and `errors` is the total number of
	/// individual index compaction failures across all batches.
	/// Resume index builds stranded by a crashed or expired owner node.
	///
	/// A `CONCURRENTLY` index build runs as a detached task. If its owning node
	/// dies mid-build, nothing waits on that generation again, so the durable
	/// build state stays in `Building`/`Closing` and the index reports
	/// `status: indexing` with a frozen counter indefinitely. This scan adopts
	/// such builds — once their owner lease has expired — via the existing
	/// expired-lease takeover (see [`IndexBuilder::resume_stalled`]) and drives
	/// them to completion.
	///
	/// The scan is lease-guarded so a single node runs it per cluster; the
	/// per-index takeover is additionally CAS-guarded, so correctness does not
	/// depend on the lease. Enumeration walks the catalog, so cost is
	/// proportional to the total index count; operators who prefer to recover
	/// stalled builds manually (with `REBUILD INDEX`) can disable the scan by
	/// setting its interval to zero.
	///
	/// Recovery is serialized: a pass adopts at most one stalled build, and
	/// no pass adopts anything while a local builder task is still running.
	/// A restart with several stranded builds (for example multiple full-text
	/// indexes on a small instance) therefore rebuilds them one at a time
	/// instead of multiplying the node's memory and CPU footprint; every
	/// stalled build is still recovered, one scan pass after the previous one
	/// finishes. User-initiated `REBUILD INDEX` is not throttled by this.
	///
	/// Returns the number of stalled builds adopted this pass.
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(self, canceller))]
	pub async fn resume_stalled_index_builds(
		&self,
		interval: Duration,
		canceller: CancellationToken,
	) -> Result<usize> {
		Self::ensure_not_cancelled(&canceller)?;
		// Single-node-per-cluster guard. The lease lasts two intervals so an
		// in-flight scan isn't preempted between ticks.
		let lh = LeaseHandler::new_with_canceller(
			self.sequences.clone(),
			self.id,
			self.transaction_factory.clone(),
			TaskLeaseType::IndexBuildResume,
			interval * 2,
			canceller.clone(),
		)?;
		if !lh.has_lease().await? {
			return Ok(0);
		}
		// Serialize recovery: while any local builder task is still running
		// (a previously adopted build or a user-started one), defer further
		// adoptions to a later pass. Without this, a restart with several
		// stranded builds starts them all at once, and the concurrent initial
		// scans can overwhelm a small instance's memory and CPU.
		if self.index_builder.has_unfinished_build().await {
			trace!(
				target: TARGET,
				"Deferring stalled index build adoption; a local index build is still running"
			);
			return Ok(0);
		}
		// Snapshot the (ns, db, table, index) hierarchy in a short read
		// transaction. Index definitions are cloned out so the per-index
		// build-state checks below don't hold the catalog transaction open.
		let mut candidates = Vec::new();
		{
			let txn = self.transaction(Read, Optimistic).await?;
			let res: Result<()> = async {
				for ns in txn.all_ns(None).await?.iter() {
					for db in txn.all_db(ns.namespace_id, None).await?.iter() {
						for tb in txn.all_tb(ns.namespace_id, db.database_id, None).await?.iter() {
							for ix in txn
								.all_tb_indexes(ns.namespace_id, db.database_id, &tb.name, None)
								.await?
								.iter()
							{
								// Indexes pending removal are cleared by the
								// tombstone reaper, not resumed.
								if ix.prepare_remove {
									continue;
								}
								candidates.push((
									ns.namespace_id,
									ns.name.clone(),
									db.database_id,
									db.name.clone(),
									tb.table_id,
									Arc::new(ix.clone()),
								));
							}
						}
					}
				}
				Ok(())
			}
			.await;
			let _ = txn.cancel().await;
			res?;
		}
		// Attempt a takeover for each candidate. `resume_stalled` is a cheap
		// no-op for healthy/online/live builds, so this is safe to call for
		// every index every pass.
		let index_builder = &self.index_builder;
		let mut resumed = 0;
		for (ns_id, ns_name, db_id, db_name, tb_id, ix) in candidates {
			Self::ensure_not_cancelled(&canceller)?;
			lh.try_maintain_lease().await?;
			let ctx = self.setup_ctx()?.freeze();
			let opt = self.setup_options(
				&Session::owner().with_ns(ns_name.as_str()).with_db(db_name.as_str()),
			);
			match index_builder
				.resume_stalled(&ctx, opt, ns_id, db_id, tb_id, Arc::clone(&ix))
				.await
			{
				Ok(true) => {
					resumed += 1;
					info!(
						target: TARGET,
						"Resuming stalled index build '{}' on table '{}'",
						ix.name, ix.table_name
					);
					// One adoption per pass: a later pass adopts the next
					// stalled build once this one has finished, keeping
					// recovery sequential on this node.
					break;
				}
				Ok(false) => {}
				Err(e) => {
					warn!(
						target: TARGET,
						"Failed to resume stalled index build '{}' on table '{}': {e}",
						ix.name, ix.table_name
					);
				}
			}
		}
		Ok(resumed)
	}

	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(dbs, canceller))]
	pub async fn index_compaction(
		dbs: Arc<Datastore>,
		interval: Duration,
		canceller: CancellationToken,
	) -> Result<(usize, usize)> {
		// Output function invocation details to logs
		trace!(target: TARGET, "Attempting index compaction process");
		// Create a new lease handler
		let lh = LeaseHandler::new_with_canceller(
			dbs.sequences.clone(),
			dbs.id,
			dbs.transaction_factory.clone(),
			TaskLeaseType::IndexCompaction,
			interval * 2,
			canceller.clone(),
		)?;
		let mut count_iteration = 0;
		let mut count_error = 0;
		// We continue without interruptions while there are keys and the lease
		'compaction: loop {
			Self::ensure_not_cancelled(&canceller)?;
			// Attempt to acquire a lease for the IndexCompaction task
			// If we don't get the lease, another node is handling this task
			if !lh.has_lease().await? {
				return Ok((count_iteration, count_error));
			}
			Self::ensure_not_cancelled(&canceller)?;
			// Output function invocation details to logs
			trace!(target: TARGET, "Running index compaction process");
			// Read the compaction queue in a short-lived read transaction
			// to avoid holding a write lock across the entire compaction cycle
			let (beg, end) = IndexCompactionKey::range();
			let range = beg..end;
			let items = {
				let txn = dbs.transaction(Read, Optimistic).await?;
				let res = txn.getr(range, None).await;
				let _ = txn.cancel().await;
				res?
			};
			Self::ensure_not_cancelled(&canceller)?;
			if items.is_empty() {
				return Ok((count_iteration, count_error));
			}
			// Collect the keys so we can delete them after processing
			let keys: Vec<Key> = items.iter().map(|(k, _)| k.clone()).collect();
			// Process compaction for each index
			count_iteration += 1;
			count_error +=
				Self::index_compaction_loop(Arc::clone(&dbs), &lh, items, canceller.clone())
					.await?;
			// Delete the processed queue entries in a separate write
			// transaction. This avoids conflicts with concurrent user
			// transactions that may enqueue new compaction requests.
			// Failed indexes are not re-enqueued here; the next user
			// write to the affected index will naturally trigger a new
			// compaction request.
			loop {
				let txn = dbs.transaction(Write, Optimistic).await?;
				if let Err(e) = Self::ensure_not_cancelled(&canceller) {
					let _ = txn.cancel().await;
					return Err(e);
				}
				for k in &keys {
					if let Err(e) = txn.del(k).await {
						warn!(target: TARGET, "Failed to delete compaction queue entry: {e}");
					}
				}
				if let Err(e) = Self::ensure_not_cancelled(&canceller) {
					let _ = txn.cancel().await;
					return Err(e);
				}
				#[cfg(test)]
				if let Err(e) = maybe_inject_retryable_conflict(
					RetryableConflictSite::IndexCompactionQueueCleanup,
					dbs.id,
				) {
					if Self::cancel_and_retry_index_operation_conflict(
						&txn,
						&e,
						"Retryable conflict committing compaction queue cleanup, retrying",
					)
					.await
					{
						continue;
					}
					warn!(target: TARGET, "Failed to commit compaction queue cleanup: {e}");
					break 'compaction;
				}
				if let Err(e) = txn.commit().await {
					if Self::cancel_and_retry_index_operation_conflict(
						&txn,
						&e,
						"Retryable conflict committing compaction queue cleanup, retrying",
					)
					.await
					{
						continue;
					}
					warn!(target: TARGET, "Failed to commit compaction queue cleanup: {e}");
					break 'compaction;
				}
				break;
			}
		}
		Ok((count_iteration, count_error))
	}

	/// Periodically drains the background reclaim queue (`/!rc` keys),
	/// destroying the data prefix of namespaces/databases/indexes that were
	/// removed by a committed `REMOVE` statement.
	///
	/// `REMOVE NAMESPACE/DATABASE/INDEX` delete only the catalog definition
	/// inside the user transaction and enqueue a [`ReclaimKey`]; this task
	/// performs the expensive data deletion out-of-band so the user statement
	/// returns immediately and never trips a request/transaction timeout.
	///
	/// Coordinated across the cluster by a [`TaskLeaseType::ReclaimTombstones`]
	/// lease so only one node reclaims at a time. The queue is read in a
	/// short-lived read transaction; each prefix is destroyed independently
	/// (idempotently — re-running on an already-empty prefix is a no-op) and the
	/// processed queue entry is then deleted in a separate write transaction, so
	/// a crash mid-reclaim simply retries on the next pass.
	///
	/// Snapshot safety: a queued removal is only reclaimed once it has been
	/// *observed* by this task for at least `grace`. The reclaim task destroys
	/// data out-of-band — on TiKV via `unsafe_destroy_range`, which bypasses
	/// MVCC — so a read transaction whose snapshot predates the `REMOVE` must be
	/// given time to finish first. On first sight the task stamps the entry's
	/// [`ReclaimState::observed_ms`]; aging is measured from there, not from the
	/// key's `uid` (a UUIDv7 stamped while the `REMOVE` statement runs, which can
	/// be arbitrarily earlier than the commit inside a long `BEGIN`/`COMMIT`
	/// block). Because the task only ever reads committed entries, `observed_ms`
	/// is always at or after the removal's commit, so `age >= grace` implies the
	/// commit is at least `grace` old — equivalent to it having fallen behind the
	/// MVCC GC safepoint (`now - tikv_gc_lifetime`) when `grace >= tikv_gc_lifetime`,
	/// by which point any transaction that could still read the data has expired.
	/// A `grace` of zero reclaims immediately without stamping (used by tests and
	/// the language-test harness teardown, where there are no concurrent readers).
	///
	/// Returns `(iterations, errors)`.
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(dbs, canceller))]
	pub async fn reclaim_tombstones(
		dbs: Arc<Datastore>,
		interval: Duration,
		grace: Duration,
		canceller: CancellationToken,
	) -> Result<(usize, usize)> {
		trace!(target: TARGET, "Attempting tombstone reclaim process");
		// Coordinate across the cluster so only one node reclaims at a time.
		let lh = LeaseHandler::new_with_canceller(
			dbs.sequences.clone(),
			dbs.id,
			dbs.transaction_factory.clone(),
			TaskLeaseType::ReclaimTombstones,
			interval * 2,
			canceller.clone(),
		)?;
		let mut count_iteration = 0;
		let mut count_error = 0;
		// Continue without interruptions while there are entries and the lease.
		loop {
			Self::ensure_not_cancelled(&canceller)?;
			// If we don't hold the lease, another node is handling this task.
			if !lh.has_lease().await? {
				return Ok((count_iteration, count_error));
			}
			Self::ensure_not_cancelled(&canceller)?;
			// Read the reclaim queue in a short-lived read transaction to avoid
			// holding a write lock across the entire reclaim cycle.
			let (beg, end) = ReclaimKey::range();
			let items = {
				let txn = dbs.transaction(Read, Optimistic).await?;
				let res = txn.getr(beg..end, None).await;
				let _ = txn.cancel().await;
				res?
			};
			Self::ensure_not_cancelled(&canceller)?;
			if items.is_empty() {
				return Ok((count_iteration, count_error));
			}
			count_iteration += 1;
			// Current wall-clock time (unix millis), used to age each entry
			// against `grace` and to stamp first observations.
			let now_ms = web_time::SystemTime::now()
				.duration_since(web_time::SystemTime::UNIX_EPOCH)
				.map(|d| d.as_millis() as u64)
				.unwrap_or(0);
			let grace_ms = grace.as_millis() as u64;
			// `done`: entries reclaimed this pass (to delete from the queue).
			// `to_stamp`: entries seen for the first time (to record an
			// observation time so aging is measured from commit, not the
			// pre-commit `uid`). Entries seen but still inside the grace window
			// are left untouched for a later pass.
			let mut done: Vec<Key> = Vec::with_capacity(items.len());
			let mut to_stamp: Vec<Key> = Vec::new();
			for (k, v) in &items {
				Self::ensure_not_cancelled(&canceller)?;
				lh.try_maintain_lease().await?;
				let rc = match ReclaimKey::decode_key(k) {
					Ok(rc) => rc,
					Err(e) => {
						count_error += 1;
						warn!(target: TARGET, "Failed to decode reclaim queue entry: {e}");
						continue;
					}
				};
				// Zero grace: reclaim immediately without the observe-then-age
				// dance (tests / harness teardown — no concurrent readers).
				if grace.is_zero() {
					match dbs.reclaim_decoded(&rc).await {
						Ok(()) => done.push(k.clone()),
						Err(e) => {
							count_error += 1;
							warn!(target: TARGET, "Tombstone reclaim failed for a queue entry: {e}");
						}
					}
					continue;
				}
				let state = match ReclaimState::kv_decode_value(v, ()) {
					Ok(s) => s,
					Err(e) => {
						count_error += 1;
						warn!(target: TARGET, "Failed to decode reclaim queue state: {e}");
						continue;
					}
				};
				if state.observed_ms == 0 {
					// First sighting of this committed entry — record an
					// observation time (>= commit) and defer reclaim. Aging is
					// measured from here, never from the pre-commit `uid`.
					to_stamp.push(k.clone());
					continue;
				}
				// Snapshot-safety gate: only destroy data once it has been
				// observed for at least `grace`, by which point any transaction
				// that could still read it is behind the MVCC GC safepoint.
				if now_ms.saturating_sub(state.observed_ms) >= grace_ms {
					match dbs.reclaim_decoded(&rc).await {
						Ok(()) => done.push(k.clone()),
						Err(e) => {
							count_error += 1;
							warn!(target: TARGET, "Tombstone reclaim failed for a queue entry: {e}");
						}
					}
				}
			}
			// Persist queue changes in a single write transaction: stamp
			// newly-observed entries and delete reclaimed ones. Done separately
			// from the read scan so we don't conflict with concurrent enqueues.
			if !done.is_empty() || !to_stamp.is_empty() {
				let txn = dbs.transaction(Write, Optimistic).await?;
				if let Err(e) = Self::ensure_not_cancelled(&canceller) {
					let _ = txn.cancel().await;
					return Err(e);
				}
				for k in &to_stamp {
					if let Ok(rc) = ReclaimKey::decode_key(k) {
						let state = ReclaimState {
							observed_ms: now_ms,
						};
						if let Err(e) = txn.set(&rc, &state).await {
							warn!(target: TARGET, "Failed to stamp reclaim queue entry: {e}");
						}
					}
				}
				for k in &done {
					if let Err(e) = txn.del(k).await {
						warn!(target: TARGET, "Failed to delete reclaim queue entry: {e}");
					}
				}
				if let Err(e) = txn.commit().await {
					let _ = txn.cancel().await;
					warn!(target: TARGET, "Failed to commit reclaim queue update: {e}");
					return Ok((count_iteration, count_error));
				}
			}
			// If this pass reclaimed nothing, the remaining entries are still
			// inside their grace window (or were just stamped, or are
			// persistently failing) — stop now and let the next timer tick retry
			// once they have aged, rather than busy-looping.
			if done.is_empty() {
				return Ok((count_iteration, count_error));
			}
		}
	}

	/// Destroy the data prefix of a decoded reclaim queue entry.
	async fn reclaim_decoded(&self, rc: &ReclaimKey<'_>) -> Result<()> {
		let expunge = rc.expunge != 0;
		match rc.kind {
			RECLAIM_NAMESPACE => {
				let prefix = crate::key::namespace::all::new(rc.ns);
				self.reclaim_prefix(&prefix, expunge).await
			}
			RECLAIM_DATABASE => {
				let prefix = crate::key::database::all::new(rc.ns, rc.db);
				self.reclaim_prefix(&prefix, expunge).await
			}
			RECLAIM_INDEX => {
				let prefix = crate::key::index::all::new(rc.ns, rc.db, rc.tb.as_ref(), rc.ix);
				self.reclaim_prefix(&prefix, expunge).await
			}
			other => {
				warn!(target: TARGET, "Unknown reclaim queue entry kind {other}, skipping");
				Ok(())
			}
		}
	}

	/// Destroy a key prefix out-of-band: `unsafe_destroy_range` on TiKV, a
	/// transactional prefix delete on every other backend. Idempotent — a
	/// re-run on an already-empty prefix is a no-op.
	async fn reclaim_prefix<K>(&self, prefix: &K, expunge: bool) -> Result<()>
	where
		K: crate::kvs::KVKey + std::fmt::Debug,
	{
		#[cfg(feature = "kv-tikv")]
		if self.tikv_ops().is_some() {
			// `unsafe_destroy_range` hard-removes every version in the range
			// in a single out-of-transaction call, so it ignores `expunge`
			// (the catalog entry is already gone — there is nothing to retain).
			let range = crate::kvs::util::to_prefix_range(prefix)?;
			return self.unsafe_destroy_range(range.start, range.end).await;
		}
		// Non-TiKV backends: delete the prefix transactionally. This runs off
		// the user request path, so even a large prefix delete here cannot trip
		// a client deadline.
		let txn = self.transaction(Write, Optimistic).await?;
		let res = if expunge {
			txn.clrp(prefix).await
		} else {
			txn.delp(prefix).await
		};
		match res {
			Ok(()) => txn.commit().await,
			Err(e) => {
				let _ = txn.cancel().await;
				Err(e)
			}
		}
	}

	#[cfg(not(target_family = "wasm"))]
	async fn await_index_compaction_handle(
		ikb: &IndexKeyBase,
		handle: &mut tokio::task::JoinHandle<Result<()>>,
		canceller: &CancellationToken,
	) {
		match handle.await {
			Ok(Ok(())) => {}
			Ok(Err(e))
				if canceller.is_cancelled()
					&& matches!(e.downcast_ref::<Error>(), Some(Error::QueryCancelled)) => {}
			Ok(Err(e)) => {
				warn!("Index compaction {ikb} fails while awaiting cancellation: {e}");
			}
			Err(e) => {
				warn!("Index compaction {ikb} join fails while awaiting cancellation: {e}");
			}
		}
	}

	#[cfg(not(target_family = "wasm"))]
	async fn await_index_compaction_handles(
		handles: &mut Vec<(IndexKeyBase, tokio::task::JoinHandle<Result<()>>)>,
		canceller: &CancellationToken,
	) {
		while let Some((ikb, mut handle)) = handles.pop() {
			Self::await_index_compaction_handle(&ikb, &mut handle, canceller).await;
		}
	}

	/// Compacts each distinct index found in the queue items.
	///
	/// On native targets, compaction tasks are spawned in parallel — one per
	/// distinct index — and joined afterwards. Duplicate queue entries for
	/// the same index are deduplicated via a [`HashMap`] so only one task is
	/// spawned per index. Failures are logged but do not abort the loop.
	///
	/// Returns the number of indexes that failed to compact.
	#[cfg(not(target_family = "wasm"))]
	async fn index_compaction_loop(
		dbs: Arc<Datastore>,
		lh: &LeaseHandler,
		items: Vec<(Key, Val)>,
		canceller: CancellationToken,
	) -> Result<usize> {
		let mut compacted_indexes = HashMap::new();
		for (k, _) in items {
			Self::ensure_not_cancelled(&canceller)?;
			lh.try_maintain_lease().await?;
			let ic = IndexCompactionKey::decode_key(&k)?;
			let ikb = IndexKeyBase::new(ic.ns, ic.db, ic.tb.as_ref().clone(), ic.ix);
			if let Entry::Vacant(e) = compacted_indexes.entry(ikb) {
				e.insert(());
			}
		}
		let mut error_count = 0;
		let mut handles: Vec<(IndexKeyBase, tokio::task::JoinHandle<Result<()>>)> =
			Vec::with_capacity(compacted_indexes.len());
		for (ikb, _) in compacted_indexes {
			if let Err(e) = Self::ensure_not_cancelled(&canceller) {
				Self::await_index_compaction_handles(&mut handles, &canceller).await;
				return Err(e);
			}
			let dbs = Arc::clone(&dbs);
			let canceller = canceller.clone();
			let task_ikb = ikb.clone();
			let jh = spawn(async move { dbs.process_index_compaction(&task_ikb, canceller).await });
			handles.push((ikb, jh));
		}
		while let Some((ikb, mut jh)) = handles.pop() {
			let res = tokio::select! {
				biased;
				_ = canceller.cancelled() => {
					Self::await_index_compaction_handle(&ikb, &mut jh, &canceller).await;
					Self::await_index_compaction_handles(&mut handles, &canceller).await;
					bail!(Error::QueryCancelled);
				}
				res = &mut jh => res?,
			};
			if let Err(e) = res {
				if canceller.is_cancelled() {
					Self::await_index_compaction_handles(&mut handles, &canceller).await;
					return Err(e);
				}
				error_count += 1;
				warn!("Index compaction {ikb} fails: {e}");
			}
		}
		Ok(error_count)
	}

	/// Compacts each distinct index found in the queue items.
	///
	/// On wasm, `tokio::spawn` is unavailable so compactions run
	/// sequentially. A [`HashSet`] is used to skip duplicate queue entries
	/// for the same index. Failures are logged but do not abort the loop,
	/// matching the non-wasm behavior so that a single transient failure
	/// does not prevent other indexes from being compacted.
	///
	/// Returns the number of indexes that failed to compact.
	#[cfg(target_family = "wasm")]
	async fn index_compaction_loop(
		dbs: Arc<Datastore>,
		lh: &LeaseHandler,
		items: Vec<(Key, Val)>,
		canceller: CancellationToken,
	) -> Result<usize> {
		let mut seen = HashSet::new();
		let mut error_count = 0;
		for (k, _) in items {
			Self::ensure_not_cancelled(&canceller)?;
			lh.try_maintain_lease().await?;
			let ic = IndexCompactionKey::decode_key(&k)?;
			let ikb = IndexKeyBase::new(ic.ns, ic.db, ic.tb.as_ref().clone(), ic.ix);
			if !seen.insert(ikb.clone()) {
				continue;
			}
			let res: Result<()> =
				async { dbs.process_index_compaction(&ikb, canceller.clone()).await }.await;
			if let Err(e) = res {
				if canceller.is_cancelled() {
					return Err(e);
				}
				error_count += 1;
				warn!("Index compaction {ikb} fails: {e}");
			}
		}
		Ok(error_count)
	}

	/// Performs the actual compaction of a single index.
	///
	/// Looks up the index definition identified by `ikb` and dispatches to
	/// the appropriate compaction implementation based on the index type:
	/// full-text, count, HNSW, or DiskANN. Indexes that are being removed
	/// (`prepare_remove`), not found, or of an unsupported type are silently
	/// skipped with a trace log.
	async fn process_index_compaction(
		&self,
		ikb: &IndexKeyBase,
		canceller: CancellationToken,
	) -> Result<()> {
		Self::ensure_not_cancelled(&canceller)?;
		let ix = {
			let txn = self.transaction(Read, Optimistic).await?;
			let res =
				txn.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None).await;
			let _ = txn.cancel().await;
			res?
		};
		Self::ensure_not_cancelled(&canceller)?;
		match ix {
			Some(ix) if !ix.prepare_remove => match &ix.index {
				Index::FullText(p) => {
					self.process_fulltext_compaction(ikb, p, &canceller).await?;
				}
				Index::Count(_) => {
					self.process_count_compaction(ikb, &canceller).await?;
				}
				Index::Hnsw(_) => {
					// HNSW compaction owns its pending-key allocation and pending-range
					// drain semantics separately from full-text/count compaction.
					self.process_hnsw_compaction(ikb, &canceller).await?;
				}
				#[cfg(diskann)]
				Index::DiskAnn(_) => {
					self.process_diskann_compaction(ikb, &canceller).await?;
				}
				_ => {
					trace!(target: TARGET, "Index compaction: Index {:?} does not support compaction, skipping", ikb);
				}
			},
			_ => {
				trace!(target: TARGET, "Index compaction: Index {:?} not found, skipping", ikb);
			}
		}
		Ok(())
	}

	/// Runs HNSW compaction as bounded read-plan/write-apply batches.
	///
	/// Pending entries are captured in a read transaction and conditionally
	/// deleted in a short write transaction before graph mutation. If a write
	/// fails after local graph mutation may have started, the cached HNSW index
	/// is evicted so later use reloads persisted state.
	async fn process_hnsw_compaction(
		&self,
		ikb: &IndexKeyBase,
		canceller: &CancellationToken,
	) -> Result<()> {
		loop {
			Self::ensure_not_cancelled(canceller)?;
			let prepared = {
				let txn = Arc::new(self.transaction(Read, Optimistic).await?);
				let res: Result<
					Option<(
						crate::catalog::TableId,
						crate::idx::trees::hnsw::index::HnswCompactionPlan,
					)>,
				> = async {
					let Some(tb) = txn.get_tb(ikb.ns(), ikb.db(), ikb.table(), None).await? else {
						return Ok(None);
					};
					match txn
						.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None)
						.await?
					{
						Some(ix) if !ix.prepare_remove && matches!(&ix.index, Index::Hnsw(_)) => {
							let mut ctx = self.setup_ctx()?;
							ctx.set_transaction(Arc::clone(&txn));
							let ctx = ctx.freeze();
							let plan = IndexOperation::prepare_hnsw_compaction(&ctx, ikb).await?;
							Ok(Some((tb.table_id, plan)))
						}
						_ => Ok(None),
					}
				}
				.await;
				let _ = txn.cancel().await;
				res?
			};
			let Some((tb, plan)) = prepared else {
				return Ok(());
			};
			if !plan.has_work() {
				return Ok(());
			}
			let has_more = plan.has_more();
			Self::ensure_not_cancelled(canceller)?;

			let txn = Arc::new(self.transaction(Write, Optimistic).await?);
			let res: Result<bool> = async {
				match txn
					.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None)
					.await?
				{
					Some(ix) if !ix.prepare_remove => match &ix.index {
						Index::Hnsw(p) => {
							let mut ctx = self.setup_ctx()?;
							ctx.set_transaction(Arc::clone(&txn));
							let ctx = ctx.freeze();
							IndexOperation::apply_hnsw_compaction(
								&ctx,
								&self.index_stores,
								ikb,
								&ix,
								p,
								plan,
							)
							.await
						}
						_ => Ok(false),
					},
					_ => Ok(false),
				}
			}
			.await;
			match res {
				Ok(true) => {
					if let Err(e) = Self::ensure_not_cancelled(canceller) {
						let _ = txn.cancel().await;
						if let Err(evict) =
							self.index_stores.remove_hnsw_index(tb, ikb.clone()).await
						{
							warn!(target: TARGET, "Failed to evict HNSW index after compaction cancellation: {evict}");
						}
						return Err(e);
					}
					#[cfg(test)]
					if let Err(e) = maybe_inject_retryable_conflict(
						RetryableConflictSite::HnswCompaction,
						self.id,
					) {
						let _ = txn.cancel().await;
						if let Err(evict) =
							self.index_stores.remove_hnsw_index(tb, ikb.clone()).await
						{
							warn!(target: TARGET, "Failed to evict HNSW index after compaction commit error: {evict}");
						}
						if Self::retry_index_operation_conflict(
							&e,
							format!(
								"Retryable conflict committing HNSW compaction for {ikb}, retrying"
							),
						)
						.await
						{
							continue;
						}
						return Err(e);
					}
					if let Err(e) = txn.commit().await {
						let _ = txn.cancel().await;
						if let Err(evict) =
							self.index_stores.remove_hnsw_index(tb, ikb.clone()).await
						{
							warn!(target: TARGET, "Failed to evict HNSW index after compaction commit error: {evict}");
						}
						if Self::retry_index_operation_conflict(
							&e,
							format!(
								"Retryable conflict committing HNSW compaction for {ikb}, retrying"
							),
						)
						.await
						{
							continue;
						}
						return Err(e);
					}
				}
				Ok(false) => {
					let _ = txn.cancel().await;
					return Ok(());
				}
				Err(e) => {
					let _ = txn.cancel().await;
					if let Err(evict) = self.index_stores.remove_hnsw_index(tb, ikb.clone()).await {
						warn!(target: TARGET, "Failed to evict HNSW index after compaction error: {evict}");
					}
					if Self::retry_index_operation_conflict(
						&e,
						format!("Retryable conflict applying HNSW compaction for {ikb}, retrying"),
					)
					.await
					{
						continue;
					}
					return Err(e);
				}
			}
			Self::ensure_not_cancelled(canceller)?;
			if !has_more {
				return Ok(());
			}
			// Defense-in-depth: every match arm above either commits (Ok(true))
			// or cancels (Ok(false)/Err) the tx, so by here `closed()` should
			// always be true. Catch any future regression where a `?` between
			// the match and this point bypasses finalization. No-op today.
			if !txn.closed() {
				let _ = txn.cancel().await;
			}
		}
	}

	#[cfg(diskann)]
	/// Runs DiskANN compaction as bounded read-plan/write-apply batches.
	async fn process_diskann_compaction(
		&self,
		ikb: &IndexKeyBase,
		canceller: &CancellationToken,
	) -> Result<()> {
		loop {
			Self::ensure_not_cancelled(canceller)?;
			let prepared = {
				let txn = Arc::new(self.transaction(Read, Optimistic).await?);
				let res: Result<
					Option<(
						crate::catalog::TableId,
						crate::idx::trees::diskann::index::DiskAnnCompactionPlan,
					)>,
				> = async {
					let Some(tb) = txn.get_tb(ikb.ns(), ikb.db(), ikb.table(), None).await? else {
						return Ok(None);
					};
					match txn
						.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None)
						.await?
					{
						Some(ix)
							if !ix.prepare_remove && matches!(&ix.index, Index::DiskAnn(_)) =>
						{
							let mut ctx = self.setup_ctx()?;
							ctx.set_transaction(Arc::clone(&txn));
							let ctx = ctx.freeze();
							let plan =
								IndexOperation::prepare_diskann_compaction(&ctx, ikb).await?;
							Ok(Some((tb.table_id, plan)))
						}
						_ => Ok(None),
					}
				}
				.await;
				let _ = txn.cancel().await;
				res?
			};
			let Some((_tb, plan)) = prepared else {
				return Ok(());
			};
			if !plan.requires_apply() {
				return Ok(());
			}
			let has_more = plan.has_more();
			Self::ensure_not_cancelled(canceller)?;

			let txn = Arc::new(self.transaction(Write, Optimistic).await?);
			let res: Result<bool> = async {
				match txn
					.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None)
					.await?
				{
					Some(ix) if !ix.prepare_remove => match &ix.index {
						Index::DiskAnn(p) => {
							let mut ctx = self.setup_ctx()?;
							ctx.set_transaction(Arc::clone(&txn));
							let ctx = ctx.freeze();
							IndexOperation::apply_diskann_compaction(
								&ctx,
								&self.index_stores,
								ikb,
								&ix,
								p,
								plan,
							)
							.await
						}
						_ => Ok(false),
					},
					_ => Ok(false),
				}
			}
			.await;
			// `apply_diskann_compaction` normally owns the transaction's
			// lifecycle (commits on success, cancels on apply failure while
			// holding the graph write lock — closing the #7318 race). A few
			// pre-apply paths inside `IndexOperation::apply_diskann_compaction`
			// (missing table or catalog lookup errors) can return without
			// finalizing the tx, so we add an idempotent safety net here:
			// cancel only if the tx is still open. Cancel on an already-closed
			// tx returns `TransactionFinished` and is harmlessly discarded.
			if !txn.closed() {
				let _ = txn.cancel().await;
			}
			match res {
				Ok(true) => {}
				Ok(false) => return Ok(()),
				Err(e) => return Err(e),
			}
			Self::ensure_not_cancelled(canceller)?;
			if !has_more {
				return Ok(());
			}
		}
	}

	/// Runs full-text compaction as a read-plan followed by a guarded write.
	///
	/// This avoids holding a mutable range scan over `!dc`/`!tt`; deltas
	/// committed after the read snapshot remain for a later compaction.
	async fn process_fulltext_compaction(
		&self,
		ikb: &IndexKeyBase,
		p: &crate::catalog::FullTextParams,
		canceller: &CancellationToken,
	) -> Result<()> {
		loop {
			Self::ensure_not_cancelled(canceller)?;
			let plan = {
				let txn = self.transaction(Read, Optimistic).await?;
				let res = IndexOperation::prepare_fulltext_compaction(
					&self.index_stores,
					ikb,
					&txn,
					p,
					&self.config.file_allowlist,
				)
				.await;
				let _ = txn.cancel().await;
				res?
			};
			if !plan.has_work() {
				return Ok(());
			}
			let has_more = plan.has_more();
			Self::ensure_not_cancelled(canceller)?;

			let txn = self.transaction(Write, Optimistic).await?;
			let res = async {
				match txn
					.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None)
					.await?
				{
					Some(ix) if !ix.prepare_remove => match &ix.index {
						Index::FullText(p) => {
							IndexOperation::apply_fulltext_compaction(
								&self.index_stores,
								ikb,
								&txn,
								p,
								&self.config.file_allowlist,
								plan,
							)
							.await
						}
						_ => Ok(false),
					},
					_ => Ok(false),
				}
			}
			.await;
			match res {
				Ok(true) => {
					if let Err(e) = Self::ensure_not_cancelled(canceller) {
						let _ = txn.cancel().await;
						return Err(e);
					}
					#[cfg(test)]
					if let Err(e) = maybe_inject_retryable_conflict(
						RetryableConflictSite::FullTextCompaction,
						self.id,
					) {
						if Self::cancel_and_retry_index_operation_conflict(
							&txn,
							&e,
							format!(
								"Retryable conflict committing full-text compaction for {ikb}, retrying"
							),
						)
						.await
						{
							continue;
						}
						return Err(e);
					}
					if let Err(e) = txn.commit().await {
						if Self::cancel_and_retry_index_operation_conflict(
							&txn,
							&e,
							format!(
								"Retryable conflict committing full-text compaction for {ikb}, retrying"
							),
						)
						.await
						{
							continue;
						}
						return Err(e);
					}
				}
				Ok(false) => {
					let _ = txn.cancel().await;
					return Ok(());
				}
				Err(e) => {
					let _ = txn.cancel().await;
					if Self::retry_index_operation_conflict(
						&e,
						format!(
							"Retryable conflict applying full-text compaction for {ikb}, retrying"
						),
					)
					.await
					{
						continue;
					}
					return Err(e);
				}
			}
			Self::ensure_not_cancelled(canceller)?;
			if !has_more {
				return Ok(());
			}
			// Defense-in-depth: every match arm above either commits (Ok(true))
			// or cancels (Ok(false)/Err) the tx, so by here `closed()` should
			// always be true. Catch any future regression where a `?` between
			// the match and this point bypasses finalization. No-op today.
			if !txn.closed() {
				let _ = txn.cancel().await;
			}
		}
	}

	/// Runs count-index compaction as a read-plan followed by a guarded write.
	///
	/// The write phase deletes only keys captured in the plan, so concurrent
	/// `!iu` deltas are preserved and included by later reads/compactions.
	async fn process_count_compaction(
		&self,
		ikb: &IndexKeyBase,
		canceller: &CancellationToken,
	) -> Result<()> {
		loop {
			Self::ensure_not_cancelled(canceller)?;
			let plan = {
				let txn = self.transaction(Read, Optimistic).await?;
				let res = IndexOperation::prepare_count_compaction(ikb, &txn).await;
				let _ = txn.cancel().await;
				res?
			};
			if !plan.has_work() {
				return Ok(());
			}
			let has_more = plan.has_more();
			Self::ensure_not_cancelled(canceller)?;

			let txn = self.transaction(Write, Optimistic).await?;
			let res = async {
				match txn
					.get_tb_index_by_id(ikb.ns(), ikb.db(), ikb.table(), ikb.index(), None)
					.await?
				{
					Some(ix) if !ix.prepare_remove && matches!(&ix.index, Index::Count(_)) => {
						IndexOperation::apply_count_compaction(ikb, &txn, plan).await
					}
					_ => Ok(false),
				}
			}
			.await;
			match res {
				Ok(true) => {
					if let Err(e) = Self::ensure_not_cancelled(canceller) {
						let _ = txn.cancel().await;
						return Err(e);
					}
					#[cfg(test)]
					if let Err(e) = maybe_inject_retryable_conflict(
						RetryableConflictSite::CountCompaction,
						self.id,
					) {
						if Self::cancel_and_retry_index_operation_conflict(
							&txn,
							&e,
							format!(
								"Retryable conflict committing count compaction for {ikb}, retrying"
							),
						)
						.await
						{
							continue;
						}
						return Err(e);
					}
					if let Err(e) = txn.commit().await {
						if Self::cancel_and_retry_index_operation_conflict(
							&txn,
							&e,
							format!(
								"Retryable conflict committing count compaction for {ikb}, retrying"
							),
						)
						.await
						{
							continue;
						}
						return Err(e);
					}
				}
				Ok(false) => {
					let _ = txn.cancel().await;
					return Ok(());
				}
				Err(e) => {
					let _ = txn.cancel().await;
					if Self::retry_index_operation_conflict(
						&e,
						format!("Retryable conflict applying count compaction for {ikb}, retrying"),
					)
					.await
					{
						continue;
					}
					return Err(e);
				}
			}
			Self::ensure_not_cancelled(canceller)?;
			if !has_more {
				return Ok(());
			}
		}
	}

	/// Process queued async events using a distributed lease to coordinate batches.
	/// Once a batch starts it runs to completion even if the lease expires, so
	/// brief overlap is possible.
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(self))]
	pub async fn event_processing(&self, interval: Duration) -> Result<()> {
		// Output function invocation details to logs
		trace!(target: TARGET, "Attempting event processing process");
		// Create a new lease handler
		let lh = LeaseHandler::new(
			self.sequences.clone(),
			self.id,
			self.transaction_factory.clone(),
			TaskLeaseType::EventProcessing,
			interval * 2,
		)?;
		// We continue without interruptions while there are keys and the lease
		loop {
			// Attempt to acquire a lease for the EventProcessing task
			// If we don't get the lease, another node is handling this task
			if !lh.has_lease().await? {
				return Ok(());
			}
			// Output function invocation details to logs
			trace!(target: TARGET, "Running event processing process");
			if AsyncEventRecord::process_next_events_batch(self, Some(&lh)).await? == 0 {
				// The last batch didn't have any events to process,
				// we can sleep until the next wake-up call
				return Ok(());
			}
		}
	}

	// --------------------------------------------------
	// Other functions
	// --------------------------------------------------

	/// Create a new transaction on this datastore
	///
	/// ```rust,no_run
	/// use surrealdb_core::kvs::{Datastore, TransactionType::*, LockType::*};
	/// use anyhow::Error;
	///
	/// #[tokio::main]
	/// async fn main() -> Result<(),Error> {
	///     let ds = Datastore::new("rocksdb://database.db").await?;
	///     let mut tx = ds.transaction(Write, Optimistic).await?;
	///     tx.cancel().await?;
	///     Ok(())
	/// }
	/// ```
	pub async fn transaction(&self, write: TransactionType, lock: LockType) -> Result<Transaction> {
		self.transaction_factory.transaction(write, lock, self.sequences.clone()).await
	}

	pub(crate) fn sequences(&self) -> &Sequences {
		&self.sequences
	}

	pub(crate) fn transaction_factory(&self) -> &TransactionFactory {
		&self.transaction_factory
	}

	#[cfg(test)]
	#[cfg_attr(not(feature = "kv-mem"), allow(dead_code))]
	pub(crate) fn index_builder(&self) -> &IndexBuilder {
		&self.index_builder
	}
	pub fn async_event_trigger(&self) -> &Arc<Notify> {
		&self.async_event_trigger
	}

	pub async fn health_check(&self) -> Result<()> {
		let tx = self.transaction(Read, Optimistic).await?;

		// Cancel the transaction
		trace!("Cancelling health check transaction");
		// Attempt to fetch data
		match tx.get(&vec![0x00], None).await {
			Err(err) => {
				// Ensure the transaction is cancelled
				let _ = tx.cancel().await;
				// Return an error for this endpoint
				Err(err)
			}
			Ok(_) => {
				// Ensure the transaction is cancelled
				let _ = tx.cancel().await;
				// Return success for this endpoint
				Ok(())
			}
		}
	}

	/// Returns how long ago the current node last refreshed its cluster
	/// heartbeat.
	///
	/// The node-membership refresh background task rewrites this node's
	/// heartbeat to the KV store every `node_membership_refresh_interval`. A
	/// small age therefore confirms, from work the node already performs, that
	/// the refresh task is running and that both the storage write path (the
	/// heartbeat was recently committed) and read path (it is readable here in
	/// a fresh transaction) are healthy. Readiness probes use this instead of a
	/// dedicated health check to avoid duplicating that work.
	pub async fn node_heartbeat_age(&self) -> Result<Duration> {
		// Open a fresh read transaction so the lookup misses the per-transaction
		// cache and actually reads the node key from storage.
		let tx = self.transaction(Read, Optimistic).await?;
		let res = tx.get_node(self.id).await;
		// Always release the transaction, regardless of the lookup result.
		let _ = tx.cancel().await;
		let node = res?;
		// Heartbeats are stored as milliseconds since the epoch; saturate to
		// avoid underflow from minor clock skew between writes and this read.
		let now = self.clock_now();
		Ok(Duration::from_millis(now.value.saturating_sub(node.heartbeat.value)))
	}

	/// Parse and execute an SQL query
	///
	/// ```rust,no_run
	/// use anyhow::Error;
	/// use surrealdb_core::kvs::Datastore;
	/// use surrealdb_core::dbs::Session;
	///
	/// #[tokio::main]
	/// async fn main() -> Result<(),Error> {
	///     let ds = Datastore::new("memory").await?;
	///     let ses = Session::owner();
	///     let ast = "USE NS test DB test; SELECT * FROM person;";
	///     let res = ds.execute(ast, &ses, None).await?;
	///     Ok(())
	/// }
	/// ```
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn execute(
		&self,
		txt: &str,
		sess: &Session,
		vars: Option<PublicVariables>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		// Parse the SQL query text
		let ast = syn::parse_with_capabilities(txt, &self.capabilities, &self.config)
			.map_err(|e| TypesError::validation(e.to_string(), None))?;
		// Process the AST
		self.process(ast, sess, vars).await
	}

	/// Parse and lower a GQL query into a [`PreparedGqlQuery`], checking that
	/// the `gql` experimental capability is enabled.
	#[cfg(feature = "gql")]
	pub(crate) fn parse_gql(&self, txt: &str) -> std::result::Result<PreparedGqlQuery, TypesError> {
		// Check if the experimental GQL capability is enabled. The
		// wording deliberately matches the existing experimental-gate errors
		// (`surrealism`, `files`) rather than naming the server's
		// `--allow-experimental` flag: core is also used embedded, where the
		// capability is enabled programmatically and no CLI flag exists.
		if !self.capabilities.allows_experimental(&ExperimentalTarget::Gql) {
			return Err(TypesError::not_allowed(
				"Experimental capability `gql` is not enabled".to_string(),
				None,
			));
		}
		// Parse and lower the GQL query text
		crate::gql::parse_with_capabilities(txt, &self.capabilities, &self.config)
			.map_err(|e| TypesError::validation(e.to_string(), None))
	}

	/// Parse and execute a GQL query
	#[cfg(feature = "gql")]
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn execute_gql(
		&self,
		txt: &str,
		sess: &Session,
		vars: Option<PublicVariables>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		// Parse and lower the GQL query text
		let plan = self.parse_gql(txt)?;
		// Process the lowered plan
		self.process_gql(plan, sess, vars).await
	}

	/// Execute a pre-lowered GQL query.
	///
	/// Mirrors [`Self::process`] for the GQL dialect: the
	/// [`PreparedGqlQuery`] already wraps a [`LogicalPlan`], so it is handed
	/// directly to the shared plan-level executor.
	#[cfg(feature = "gql")]
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn process_gql(
		&self,
		q: PreparedGqlQuery,
		sess: &Session,
		vars: Option<PublicVariables>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_plan_inner(q.0, sess, vars, None).await
	}

	/// Execute a pre-lowered GQL query with an externally-owned cancellation
	/// handle. See [`Self::process_with_transaction_and_cancel`] for the
	/// cancellation semantics.
	#[cfg(feature = "gql")]
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub(crate) async fn process_gql_with_cancel(
		&self,
		q: PreparedGqlQuery,
		sess: &Session,
		vars: Option<PublicVariables>,
		cancel: CancelHandle,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_plan_inner(q.0, sess, vars, Some(cancel)).await
	}

	/// Execute a pre-lowered GQL query with an existing transaction.
	#[cfg(feature = "gql")]
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub(crate) async fn process_gql_with_transaction(
		&self,
		q: PreparedGqlQuery,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_plan_with_transaction_inner(q.0, sess, vars, tx, None).await
	}

	/// Execute a pre-lowered GQL query with an existing transaction and an
	/// externally-owned cancellation handle. See
	/// [`Self::process_with_transaction_and_cancel`] for the cancellation
	/// semantics.
	#[cfg(feature = "gql")]
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub(crate) async fn process_gql_with_transaction_and_cancel(
		&self,
		q: PreparedGqlQuery,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
		cancel: CancelHandle,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_plan_with_transaction_inner(q.0, sess, vars, tx, Some(cancel)).await
	}

	/// Execute a SurrealQL query with an externally-owned cancellation
	/// handle. See [`Self::process_with_transaction_and_cancel`] for the
	/// cancellation semantics.
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn execute_with_cancel(
		&self,
		txt: &str,
		sess: &Session,
		vars: Option<PublicVariables>,
		cancel: CancelHandle,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		// Parse the SQL query text
		let ast = syn::parse_with_capabilities(txt, &self.capabilities, &self.config)
			.map_err(|e| TypesError::validation(e.to_string(), None))?;
		// Process the AST
		self.process_with_cancel(ast, sess, vars, cancel).await
	}

	/// Execute a query with an existing transaction
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn execute_with_transaction(
		&self,
		txt: &str,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		// Parse the SQL query text
		let ast = syn::parse_with_capabilities(txt, &self.capabilities, &self.config)
			.map_err(|e| TypesError::validation(e.to_string(), None))?;
		// Process the AST with the transaction
		self.process_with_transaction(ast, sess, vars, tx).await
	}

	/// Execute a query with an existing transaction and an externally-owned
	/// cancellation handle. See [`Self::process_with_transaction_and_cancel`]
	/// for the cancellation semantics.
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn execute_with_transaction_and_cancel(
		&self,
		txt: &str,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
		cancel: CancelHandle,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		// Parse the SQL query text
		let ast = syn::parse_with_capabilities(txt, &self.capabilities, &self.config)
			.map_err(|e| TypesError::validation(e.to_string(), None))?;
		// Process the AST with the transaction
		self.process_with_transaction_and_cancel(ast, sess, vars, tx, cancel).await
	}

	/// Process an AST with an existing transaction
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn process_with_transaction(
		&self,
		ast: Ast,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_with_transaction_inner(ast, sess, vars, tx, None).await
	}

	/// Process an AST with an existing transaction and an externally-owned
	/// cancellation handle. When the handle is tripped the executor's
	/// `Context::done` walks short-circuit with `Reason::Canceled` at the
	/// next yield point (statement boundary, iterator hot loop check, etc.)
	/// and the transaction is finalised on the executor's normal error path.
	/// Bare-await sites (e.g. `SLEEP`) `select!` against the handle's
	/// awaitable view, so they are also interrupted instead of running to
	/// completion.
	///
	/// Used by the WebSocket RPC layer so a client disconnect cancels the
	/// in-flight query without waiting for it to run to completion. The
	/// caller owns the [`CancelHandle`] and is responsible for tripping it.
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn process_with_transaction_and_cancel(
		&self,
		ast: Ast,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
		cancel: CancelHandle,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_with_transaction_inner(ast, sess, vars, tx, Some(cancel)).await
	}

	/// Ast-level shim over [`Self::process_plan_with_transaction_inner`]: the
	/// SurrealQL surface AST is converted to a [`LogicalPlan`] before the
	/// shared plan-level body runs. Keeps the SurrealQL `Ast` entry points
	/// unchanged while letting GQL (which already lowers to a `LogicalPlan`)
	/// share the same execution path.
	async fn process_with_transaction_inner(
		&self,
		ast: Ast,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
		cancel: Option<CancelHandle>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_plan_with_transaction_inner(ast.into(), sess, vars, tx, cancel).await
	}

	/// Shared body for [`Self::process_with_transaction`],
	/// [`Self::process_with_transaction_and_cancel`] and the GQL
	/// `process_gql` variants. The public Ast variants exist to keep the
	/// non-cancel API stable for SDK / embedded callers; `cancel: None`
	/// reproduces the pre-cancellation behaviour exactly.
	async fn process_plan_with_transaction_inner(
		&self,
		plan: LogicalPlan,
		sess: &Session,
		vars: Option<PublicVariables>,
		tx: Arc<Transaction>,
		cancel: Option<CancelHandle>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		// Check if the session has expired
		if sess.expired() {
			return Err(TypesError::not_allowed(
				"The session has expired".to_string(),
				AuthError::SessionExpired,
			));
		}

		// Check if anonymous actors can execute queries when auth is enabled
		if let Err(e) = self.check_anon(sess) {
			return Err(TypesError::not_allowed(
				format!("Anonymous access not allowed: {e}"),
				AuthError::NotAllowed {
					actor: "anonymous".to_owned(),
					action: "process".to_owned(),
					resource: "query".to_owned(),
				},
			));
		}

		// Create a new query options
		let opt = self.setup_options(sess);

		// Create a default context
		let mut ctx = self.setup_ctx().map_err(|e| {
			e.downcast::<Error>()
				.map(crate::err::into_types_error)
				.unwrap_or_else(|e| TypesError::internal(e.to_string()))
		})?;

		// Install the external cancellation handle if one was supplied.
		// The executor checks `Context::done` between statements and in
		// iterator hot loops, so flipping the flag mid-query causes the
		// next yield to return `Error::QueryCancelled` and the executor's
		// error path finalises the transaction cleanly. Bare-await sites
		// (`SLEEP`) `select!` against the handle's awaitable view.
		if let Some(cancel) = cancel {
			ctx.set_cancellation(&cancel);
		}

		// Start an execution context
		ctx.attach_session(sess).map_err(crate::err::into_types_error)?;

		// Store the query variables
		if let Some(vars) = vars {
			ctx.attach_variables(vars.into()).map_err(crate::err::into_types_error)?;
		}

		// Propagate the resolved tenant identity onto the externally-supplied
		// transaction so the emitted [`crate::observe::TransactionEvent`]
		// carries the active session's namespace, database, user, etc.
		if let Some(identity) = ctx.tenant_identity() {
			tx.set_tenant_identity(Arc::clone(identity));
		}

		// Set the transaction in the context
		ctx.set_transaction(tx);

		// Process all statements with the transaction
		Executor::execute_plan_with_transaction(self, ctx.freeze(), opt, plan).await.map_err(|e| {
			e.downcast::<Error>()
				.map(crate::err::into_types_error)
				.unwrap_or_else(|e| TypesError::internal(e.to_string()))
		})
	}

	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn execute_import<S>(
		&self,
		sess: &Session,
		vars: Option<PublicVariables>,
		query: S,
	) -> Result<Vec<QueryResult>>
	where
		S: Stream<Item = Result<Bytes>>,
	{
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);

		// Check if anonymous actors can execute queries when auth is enabled
		// TODO(sgirones): Check this as part of the authorisation layer
		self.check_anon(sess).map_err(|_| {
			Error::from(IamError::NotAllowed {
				actor: "anonymous".to_string(),
				action: "process".to_string(),
				resource: "query".to_string(),
			})
		})?;

		// Create a new query options
		let opt = self.setup_options(sess);

		// Create a default context
		let mut ctx = self.setup_ctx()?;
		// Start an execution context
		ctx.attach_session(sess)?;
		// Store the query variables
		if let Some(vars) = vars {
			ctx.attach_variables(vars.into())?;
		}
		// Process all statements

		let parser_settings = ParserSettings {
			files_enabled: ctx.get_capabilities().allows_experimental(&ExperimentalTarget::Files),
			surrealism_enabled: ctx
				.get_capabilities()
				.allows_experimental(&ExperimentalTarget::Surrealism),
			..Default::default()
		};
		let mut statements_stream = StatementStream::new_with_settings(parser_settings);
		let mut buffer = BytesMut::new();
		let mut parse_size = 4096;
		let mut bytes_stream = pin!(query);
		let mut complete = false;
		let mut filling = true;

		let stream = futures::stream::poll_fn(move |cx| {
			loop {
				// fill the buffer to at least parse_size when filling is required.
				while filling {
					let bytes = ready!(bytes_stream.as_mut().poll_next(cx));
					let bytes = match bytes {
						Some(Err(e)) => return Poll::Ready(Some(Err(e))),
						Some(Ok(x)) => x,
						None => {
							complete = true;
							filling = false;
							break;
						}
					};

					buffer.extend_from_slice(&bytes);
					filling = buffer.len() < parse_size
				}

				// if we finished streaming we can parse with complete so that the parser can be
				// sure of it's results.
				if complete {
					return match statements_stream.parse_complete(&mut buffer) {
						Err(e) => {
							Poll::Ready(Some(Err(anyhow::Error::new(Error::InvalidQuery(e)))))
						}
						Ok(None) => Poll::Ready(None),
						Ok(Some(x)) => Poll::Ready(Some(Ok(x))),
					};
				}

				// otherwise try to parse a single statement.
				match statements_stream.parse_partial(&mut buffer) {
					Err(e) => {
						return Poll::Ready(Some(Err(anyhow::Error::new(Error::InvalidQuery(e)))));
					}
					Ok(Some(x)) => return Poll::Ready(Some(Ok(x))),
					Ok(None) => {
						// Couldn't parse a statement for sure.
						if buffer.len() >= parse_size && parse_size < u32::MAX as usize {
							// the buffer already contained more or equal to parse_size bytes
							// this means we are trying to parse a statement of more then buffer
							// size. so we need to increase the buffer size.
							parse_size = (parse_size + 1).next_power_of_two();
						}
						// start filling the buffer again.
						filling = true;
					}
				}
			}
		});

		Executor::execute_stream(self, Arc::new(ctx), opt, true, stream).await
	}

	/// Execute a pre-parsed SQL query
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn process(
		&self,
		ast: Ast,
		sess: &Session,
		vars: Option<PublicVariables>,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		//TODO: Insert planner here.
		self.process_plan_inner(ast.into(), sess, vars, None).await
	}

	/// Execute a pre-parsed SQL query with an externally-owned cancellation
	/// handle. See [`Self::process_with_transaction_and_cancel`] for the
	/// cancellation semantics.
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn process_with_cancel(
		&self,
		ast: Ast,
		sess: &Session,
		vars: Option<PublicVariables>,
		cancel: CancelHandle,
	) -> std::result::Result<Vec<QueryResult>, TypesError> {
		self.process_plan_inner(ast.into(), sess, vars, Some(cancel)).await
	}

	pub(crate) async fn process_plan(
		&self,
		plan: LogicalPlan,
		sess: &Session,
		vars: Option<PublicVariables>,
	) -> Result<Vec<QueryResult>, TypesError> {
		self.process_plan_inner(plan, sess, vars, None).await
	}

	async fn process_plan_inner(
		&self,
		plan: LogicalPlan,
		sess: &Session,
		vars: Option<PublicVariables>,
		cancel: Option<CancelHandle>,
	) -> Result<Vec<QueryResult>, TypesError> {
		// Check if the session has expired
		if sess.expired() {
			return Err(TypesError::not_allowed(
				"The session has expired".to_string(),
				AuthError::SessionExpired,
			));
		}

		// Check if anonymous actors can execute queries when auth is enabled
		// TODO(sgirones): Check this as part of the authorisation layer
		if let Err(e) = self.check_anon(sess) {
			return Err(TypesError::not_allowed(
				format!("Anonymous access not allowed: {e}"),
				AuthError::NotAllowed {
					actor: "anonymous".to_owned(),
					action: "process".to_owned(),
					resource: "query".to_owned(),
				},
			));
		}

		// Create a new query options
		let opt = self.setup_options(sess);

		// Create a default context
		let mut ctx = self.setup_ctx().map_err(|e| {
			e.downcast::<Error>()
				.map(crate::err::into_types_error)
				.unwrap_or_else(|e| TypesError::internal(e.to_string()))
		})?;

		// Install the external cancellation flag if one was supplied. See
		// [`Self::process_with_transaction_and_cancel`].
		if let Some(cancel) = cancel {
			ctx.set_cancellation(&cancel);
		}

		// Start an execution context
		ctx.attach_session(sess).map_err(crate::err::into_types_error)?;

		// Store the query variables
		if let Some(vars) = vars {
			ctx.attach_variables(vars.into()).map_err(crate::err::into_types_error)?;
		}

		// Process all statements
		Executor::execute_plan(self, ctx.freeze(), opt, plan).await.map_err(|e| {
			e.downcast::<Error>()
				.map(crate::err::into_types_error)
				.unwrap_or_else(|e| TypesError::internal(e.to_string()))
		})
	}

	/// Evaluates a SQL [`Value`] without checking authenticating config
	/// This is used in very specific cases, where we do not need to check
	/// whether authentication is enabled, or guest access is disabled.
	/// For example, this is used when processing a record access SIGNUP or
	/// SIGNIN clause, which still needs to work without guest access.
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub(crate) async fn evaluate(
		&self,
		val: &Expr,
		sess: &Session,
		vars: Option<PublicVariables>,
	) -> Result<PublicValue> {
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);
		// Create a new memory stack
		let mut stack = TreeStack::new();
		// Create a new query options
		let opt = self.setup_options(sess);
		// Create a default context
		let mut ctx = self.setup_ctx()?;
		// Set the global query timeout
		if let Some(timeout) = self.dynamic_configuration.get_query_timeout() {
			ctx.add_timeout(timeout)?;
		}

		let txn_type = if val.read_only() {
			TransactionType::Read
		} else {
			TransactionType::Write
		};
		// Start a new transaction. Tenant identity is attached up-front so the
		// emitted [`crate::observe::TransactionEvent`] carries the session's
		// namespace, database, user, etc.
		let txn = self
			.transaction(txn_type, Optimistic)
			.await?
			.with_tenant_identity(Some(Arc::new(crate::observe::TenantIdentity::from_session(
				sess,
			))))
			.enclose();
		// Store the transaction
		ctx.set_transaction(Arc::clone(&txn));

		// Start an execution context
		ctx.attach_session(sess)?;
		// Store the query variables
		if let Some(vars) = vars {
			ctx.attach_public_variables(vars)?;
		}

		// Freeze the context
		let ctx = ctx.freeze();
		// Compute the value
		let res =
			stack.enter(|stk| val.compute(stk, &ctx, &opt, None)).finish().await.catch_return();
		// Store any data
		if res.is_ok() && txn_type == TransactionType::Write {
			// If the compute was successful, then commit if writeable
			txn.commit().await?;
		} else {
			// Cancel if the compute was an error, or if readonly
			txn.cancel().await?;
		};
		// Return result
		convert_value_to_public_value(res?)
	}

	/// Performs a database import from SQL
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn import(&self, sql: &str, sess: &Session) -> Result<Vec<QueryResult>> {
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);
		// Execute the SQL import
		self.execute(sql, sess, None).await.map_err(|e| anyhow::anyhow!(e))
	}

	/// Performs a database import from SQL
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn import_stream<S>(&self, sess: &Session, stream: S) -> Result<Vec<QueryResult>>
	where
		S: Stream<Item = Result<Bytes>>,
	{
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);
		// Execute the SQL import
		self.execute_import(sess, None, stream).await
	}

	/// Performs a full database export as SQL
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn export(
		&self,
		sess: &Session,
		chn: Sender<Vec<u8>>,
	) -> Result<impl Future<Output = Result<()>>> {
		// Create a default export config
		let cfg = super::export::Config::default();
		self.export_with_config(sess, chn, cfg).await
	}

	/// Performs a full database export as SQL
	#[instrument(level = "debug", target = "surrealdb::core::kvs::ds", skip_all)]
	pub async fn export_with_config(
		&self,
		sess: &Session,
		chn: Sender<Vec<u8>>,
		cfg: export::Config,
	) -> Result<impl Future<Output = Result<()>> + 'static> {
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);
		// Retrieve the provided NS and DB
		let (ns, db) = crate::iam::check::check_ns_db(sess)?;
		// Create a new readonly transaction
		let txn = self.transaction(Read, Optimistic).await?;
		let batch_size = self.config.export_batch_size;
		// Return an async export job
		Ok(async move {
			// Process the export
			let res = txn.export(&ns, &db, cfg, batch_size, chn).await;
			txn.cancel().await?;
			res
		})
	}

	/// Checks the required permissions level for this session
	#[instrument(level = "trace", target = "surrealdb::core::kvs::ds", skip(self, sess))]
	#[allow(clippy::needless_pass_by_value)] // Public API: ergonomic for callers passing `ResourceKind::X.on_db(ns, db)` inline.
	pub fn check(&self, sess: &Session, action: Action, resource: Resource) -> Result<()> {
		// Check if the session has expired
		ensure!(!sess.expired(), Error::ExpiredSession);
		// Skip auth for Anonymous users if auth is disabled
		let skip_auth = !self.is_auth_enabled() && sess.au.is_anon();
		if !skip_auth {
			sess.au.is_allowed(action, &resource)?;
		}
		// All ok
		Ok(())
	}

	pub fn setup_options(&self, sess: &Session) -> Options {
		Options::new(&self.config)
			.with_ns(sess.ns())
			.with_db(sess.db())
			.with_auth(Arc::clone(&sess.au))
	}

	pub fn setup_ctx(&self) -> Result<Context> {
		let ctx = Context::from_ds(
			self.id,
			self.auth_enabled,
			self.dynamic_configuration.clone(),
			self.dynamic_configuration.get_query_timeout(),
			self.slow_log.clone(),
			Arc::clone(&self.capabilities),
			self.index_stores.clone(),
			self.index_builder.clone(),
			self.sequences.clone(),
			Arc::clone(&self.cache),
			Arc::clone(&self.function_registry),
			#[cfg(feature = "http")]
			Arc::clone(&self.http_client),
			#[cfg(storage)]
			self.temporary_directory.clone(),
			self.buckets.clone(),
			Arc::clone(&self.config),
			#[cfg(feature = "surrealism")]
			Arc::clone(&self.surrealism_cache),
		)?;
		Ok(ctx)
	}

	/// check for disallowed anonymous users
	pub fn check_anon(&self, sess: &Session) -> Result<(), IamError> {
		if self.auth_enabled && sess.au.is_anon() && !self.capabilities.allows_guest_access() {
			Err(IamError::NotAllowed {
				actor: "anonymous".to_string(),
				action: String::new(),
				resource: String::new(),
			})
		} else {
			Ok(())
		}
	}

	/// SECURITY: `USE NS` implicitly creates the namespace when it does
	/// not exist. `DEFINE NAMESPACE` requires `Edit` on `Namespace`@`Root`,
	/// so the same authorization must gate the materialization step in
	/// `USE`. Returns `true` if the caller may proceed with
	/// `get_or_add_ns` — either because the namespace already exists
	/// (in which case the call is a no-op lookup) or because the caller
	/// has the necessary authorization. Returns `false` when the
	/// namespace does not exist *and* the caller lacks permission;
	/// callers that still want to set the session context for a later
	/// authenticated step (the typical pre-signin RPC `USE` pattern)
	/// can do so without triggering creation. See `SECURITY_GUIDE.md`
	/// section 3.
	pub(crate) async fn should_materialize_ns_on_use(
		&self,
		tx: &Transaction,
		auth: &Auth,
		ns: &str,
	) -> Result<bool> {
		if tx.get_ns_by_name(ns, None).await?.is_some() {
			return Ok(true);
		}
		if !self.auth_enabled && auth.is_anon() {
			return Ok(true);
		}
		Ok(auth.is_allowed(Action::Edit, &ResourceKind::Namespace.on_root()).is_ok())
	}

	/// SECURITY: counterpart to [`Self::should_materialize_ns_on_use`]
	/// for the database half of `USE`. Implicit creation requires the
	/// same authorization as `DEFINE DATABASE` (`Edit` on `Database`@`Ns`),
	/// AND the parent namespace must already exist or the caller must
	/// also be authorized to create it — `ensure_ns_db` is
	/// `get_or_add_db_upwards(..., upwards = true)` and will silently
	/// `get_or_add_ns` the parent when it is missing
	/// (`kvs/tx.rs::get_or_add_db_upwards`). Without the second check a
	/// namespace-level Editor on a stale token (the namespace was
	/// dropped after the token was issued) could recreate the parent
	/// namespace as a side effect of `USE NS dropped DB anything`.
	pub(crate) async fn should_materialize_db_on_use(
		&self,
		tx: &Transaction,
		auth: &Auth,
		ns: &str,
		db: &str,
	) -> Result<bool> {
		if tx.get_db_by_name(ns, db, None).await?.is_some() {
			return Ok(true);
		}
		if !self.auth_enabled && auth.is_anon() {
			return Ok(true);
		}
		// Block the upwards-create side effect: if the parent namespace
		// is missing and the caller can't create namespaces, refuse.
		if tx.get_ns_by_name(ns, None).await?.is_none()
			&& auth.is_allowed(Action::Edit, &ResourceKind::Namespace.on_root()).is_err()
		{
			return Ok(false);
		}
		Ok(auth.is_allowed(Action::Edit, &ResourceKind::Database.on_ns(ns)).is_ok())
	}

	pub async fn process_use(
		&self,
		ctx: Option<&Context>,
		session: &mut Session,
		namespace: Option<String>,
		database: Option<String>,
	) -> std::result::Result<QueryResult, TypesError> {
		let new_tx = || async {
			self.transaction(Write, Optimistic)
				.await
				.map_err(|err| TypesError::internal(err.to_string()))
		};
		let commit_tx = |txn: Transaction| async move {
			txn.commit().await.map_err(|err| TypesError::internal(err.to_string()))
		};

		let query_result = QueryResultBuilder::started_now();
		// SECURITY: `process_use` may be called before the caller has
		// authenticated (e.g. SDKs that call `use` before `signin`). To
		// preserve that pattern without re-opening the bypass that this
		// guard closes, we set the session context but only commit
		// implicit `DEFINE NAMESPACE` / `DEFINE DATABASE`-equivalent
		// creation when the caller has authorization for it. Callers
		// without permission end up with a session that targets a
		// resource that may not exist; downstream operations surface a
		// clean `NsNotFound` / `DbNotFound` rather than a silently
		// auto-created namespace they should not have been able to
		// create. See `SECURITY_GUIDE.md` section 3.
		let map_internal = |err: anyhow::Error| TypesError::internal(err.to_string());
		match (namespace, database) {
			(Some(ns), Some(db)) => {
				let tx = new_tx().await?;
				let create_ns = self
					.should_materialize_ns_on_use(&tx, &session.au, &ns)
					.await
					.map_err(map_internal)?;
				let create_db = create_ns
					&& self
						.should_materialize_db_on_use(&tx, &session.au, &ns, &db)
						.await
						.map_err(map_internal)?;
				if create_db {
					tx.ensure_ns_db(ctx, &ns, &db).await.map_err(map_internal)?;
					commit_tx(tx).await?;
				} else if create_ns {
					tx.get_or_add_ns(ctx, &ns).await.map_err(map_internal)?;
					commit_tx(tx).await?;
				} else {
					let _ = tx.cancel().await;
				}
				session.ns = Some(ns);
				session.db = Some(db);
			}
			(Some(ns), None) => {
				let tx = new_tx().await?;
				let create_ns = self
					.should_materialize_ns_on_use(&tx, &session.au, &ns)
					.await
					.map_err(map_internal)?;
				if create_ns {
					tx.get_or_add_ns(ctx, &ns).await.map_err(map_internal)?;
					commit_tx(tx).await?;
				} else {
					let _ = tx.cancel().await;
				}
				session.ns = Some(ns);
			}
			(None, Some(db)) => {
				let Some(ns) = session.ns.clone() else {
					return Err(TypesError::validation(
						"Cannot use database without namespace".to_string(),
						None,
					));
				};
				let tx = new_tx().await?;
				let create_db = self
					.should_materialize_db_on_use(&tx, &session.au, &ns, &db)
					.await
					.map_err(map_internal)?;
				if create_db {
					tx.ensure_ns_db(ctx, &ns, &db).await.map_err(map_internal)?;
					commit_tx(tx).await?;
				} else {
					let _ = tx.cancel().await;
				}
				session.db = Some(db);
			}
			(None, None) => {
				session.ns = None;
				session.db = None;
			}
		}

		let value = PublicValue::from_t(object! {
			namespace: session.ns.clone(),
			database: session.db.clone(),
		});

		Ok(query_result.finish_with_result(Ok(value)))
	}

	/// Get a db model by name.
	///
	/// TODO: This should not be public, but it is used by callers outside the
	/// `surrealdb-core` crate (the SDK's local engine and the server's ML route).
	pub async fn get_db_model(
		&self,
		ns: &str,
		db: &str,
		model_name: &str,
		model_version: &str,
	) -> Result<Option<Arc<crate::catalog::MlModelDefinition>>> {
		let tx = self.transaction(Read, Optimistic).await?;
		let db = tx.expect_db_by_name(ns, db).await?;
		let model = tx
			.get_db_model(db.namespace_id, db.database_id, model_name, model_version, None)
			.await?;
		tx.cancel().await?;
		Ok(model)
	}

	/// Invoke an API handler.
	///
	/// TODO: This should not need to be public, but it is used by the server's
	/// HTTP API route (outside the `surrealdb-core` crate).
	pub async fn invoke_api_handler(
		&self,
		ns: &str,
		db: &str,
		path: &str,
		session: &Session,
		mut req: ApiRequest,
	) -> Result<ApiResponse> {
		// Enforce the tenant boundary before resolving or dispatching anything.
		// The namespace/database come from caller-controlled input (the
		// `/api/:ns/:db/:endpoint` URL path) and the HTTP route has already
		// overwritten the session's selected ns/db with them, so the
		// authenticated level is the only trustworthy scope. A principal
		// authenticated for one tenant must not be able to invoke another
		// tenant's custom API — whose handler runs with permissions disabled
		// (GHSA-848m-r628-vrxw).
		if !session.au.can_access_ns_db(ns, db) {
			debug!(
				request_id = %req.request_id,
				"Custom API request denied: URL namespace/database is outside the authenticated session scope"
			);
			return Ok(ApiResponse::from_error(ApiError::PermissionDenied, req.request_id.clone()));
		}

		let tx = Arc::new(self.transaction(TransactionType::Write, LockType::Optimistic).await?);

		let db = tx.ensure_ns_db(None, ns, db).await?;

		let apis = tx.all_db_apis(db.namespace_id, db.database_id, None).await?;
		let segments: Vec<&str> = path.split('/').filter(|x| !x.is_empty()).collect();

		let res = match ApiDefinition::find_definition(apis.as_ref(), &segments, req.method) {
			Some((api, params)) => {
				debug!(
					request_id = %req.request_id,
					path = %path,
					"API definition found, dispatching to process_api_request"
				);
				req.params = params.try_into()?;

				let opt = self.setup_options(session);

				let mut ctx = self.setup_ctx()?;
				ctx.set_transaction(Arc::clone(&tx));
				ctx.attach_session(session)?;
				let ctx = &ctx.freeze();

				process_api_request(ctx, &opt, api, req).await
			}
			None => {
				trace!(
					request_id = %req.request_id,
					path = %path,
					"No API definition found for path"
				);
				tx.cancel().await?;
				return Ok(ApiResponse::from_error(ApiError::NotFound, req.request_id.clone()));
			}
		};

		// Handle committing or cancelling the transaction
		if res.is_ok() {
			tx.commit().await?;
		} else {
			tx.cancel().await?;
		}

		res
	}

	pub async fn put_ml_model(
		&self,
		session: &Session,
		name: &str,
		version: &str,
		description: &str,
		data: Vec<u8>,
	) -> Result<()> {
		let ns = session.ns.as_ref().context("Namespace is required")?;
		let db = session.db.as_ref().context("Database is required")?;

		self.check(session, Action::Edit, ResourceKind::Model.on_db(ns, db))?;

		// Calculate the hash of the model file
		let hash = crate::obs::hash(&data);
		// Calculate the path of the model file
		let path = get_model_path(ns, db, name, version, &hash);
		// Insert the file data in to the store
		crate::obs::put(&path, data).await?;
		// Insert the model in to the database
		let model = DefineModelStatement {
			name: name.to_string().into(),
			version: version.to_string().into(),
			comment: Expr::Literal(Literal::String(description.into())),
			hash: hash.into(),
			kind: Default::default(),
			permissions: Default::default(),
		};

		let q = LogicalPlan {
			expressions: vec![TopLevelExpr::Expr(Expr::Define(Box::new(DefineStatement::Model(
				model,
			))))],
		};

		self.process_plan(q, session, None).await.map_err(|e| anyhow::anyhow!(e))?;

		Ok(())
	}

	pub fn config(&self) -> Arc<CommonConfig> {
		Arc::clone(&self.config)
	}

	/// Retrieve (or generate and cache) the GraphQL schema for the namespace
	/// and database selected on `session`. Backed by the datastore-wide schema
	/// cache so every transport (HTTP `/graphql`, WebSocket subscriptions, the
	/// `graphql` RPC method, the MCP `graphql` tool) shares one set of compiled
	/// schemas and a single DDL-driven invalidation path.
	#[cfg(all(feature = "graphql", not(target_family = "wasm")))]
	pub async fn graphql_schema(
		self: &Arc<Self>,
		session: &Session,
	) -> Result<async_graphql::dynamic::Schema, crate::graphql::GraphqlError> {
		self.graphql_schema_cache.get_schema(self, session).await
	}

	#[cfg(feature = "http")]
	pub fn http_client(&self) -> Arc<HttpClient> {
		Arc::clone(&self.http_client)
	}

	/// Builds a [`NodeEndpointResolver`] backed by this datastore's catalog. Used by the
	/// builder to hand a resolver to clustered message brokers after the datastore is
	/// fully constructed.
	///
	/// Only available on non-WASM targets: the underlying transaction types are not
	/// `Send + Sync` under the WASM single-threaded model, and cross-node delivery
	/// (the only consumer) doesn't apply to in-browser datastores.
	#[cfg(not(target_family = "wasm"))]
	pub(crate) fn endpoint_resolver(&self) -> Arc<dyn crate::dbs::NodeEndpointResolver> {
		Arc::new(CatalogNodeEndpointResolver {
			transaction_factory: self.transaction_factory.clone(),
			sequences: self.sequences.clone(),
		})
	}
}

/// Catalog-backed [`NodeEndpointResolver`] handed to clustered brokers post-construction.
///
/// Holds clones of the transaction factory and sequences (both cheap `Arc`-based clones) so
/// it can open read transactions independently of the [`Datastore`] struct, avoiding any
/// reference cycle between the broker and the datastore.
#[cfg(not(target_family = "wasm"))]
#[derive(Clone)]
struct CatalogNodeEndpointResolver {
	transaction_factory: TransactionFactory,
	sequences: Sequences,
}

#[cfg(not(target_family = "wasm"))]
impl std::fmt::Debug for CatalogNodeEndpointResolver {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		f.debug_struct("CatalogNodeEndpointResolver").finish_non_exhaustive()
	}
}

#[cfg(not(target_family = "wasm"))]
impl crate::dbs::NodeEndpointResolver for CatalogNodeEndpointResolver {
	fn resolve(
		&self,
		target_node: [u8; 16],
	) -> std::pin::Pin<Box<dyn std::future::Future<Output = Option<String>> + Send + '_>> {
		Box::pin(async move {
			let uuid = Uuid::from_bytes(target_node);
			let txn = self
				.transaction_factory
				.transaction(Read, Optimistic, self.sequences.clone())
				.await
				.ok()?;
			let key = crate::key::root::nd::Nd::new(uuid);
			let node: Option<Node> = txn.get(&key, None).await.ok()?;
			let _ = txn.cancel().await;
			node.and_then(|n| n.http_endpoint)
		})
	}
}

#[cfg(test)]
mod test {
	use std::collections::BTreeMap;
	use std::future::pending;

	use super::*;
	use crate::catalog::providers::{
		CatalogProvider, DatabaseProvider, NamespaceProvider, TableProvider,
	};
	use crate::iam::verify::verify_root_creds;
	use crate::kvs::testing::{
		RetryableConflictSite, inject_retryable_conflict, retryable_conflict_count,
	};
	use crate::types::{PublicValue, PublicVariables};
	use crate::val::TableName;

	async fn new_index_compaction_test_ds() -> Result<(Datastore, Session)> {
		let ds = Datastore::new("memory").await?;
		let session = Session::owner().with_ns("test").with_db("test");
		let txn = ds.transaction(Write, Pessimistic).await?;
		txn.ensure_ns_db(None, "test", "test").await?;
		txn.commit().await?;
		Ok((ds, session))
	}

	async fn execute_all(ds: &Datastore, session: &Session, sql: &str) -> Result<()> {
		for result in ds.execute(sql, session, None).await? {
			result.result?;
		}
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn execute_gql_requires_experimental_capability() -> Result<()> {
		let ds = Datastore::new("memory").await?;
		let ses = Session::owner().with_ns("test").with_db("test");
		let err = ds.execute_gql("MATCH (n:person) RETURN n", &ses, None).await.unwrap_err();
		assert!(
			err.to_string().contains("Experimental capability `gql` is not enabled"),
			"unexpected error: {err}"
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn execute_gql_parses_lowers_and_executes() -> Result<()> {
		use crate::dbs::capabilities::Targets;
		let ds = Datastore::builder()
			.with_capabilities(Capabilities::all().with_experimental(Targets::All))
			.build_with_path("memory")
			.await?;
		let ses = Session::owner().with_ns("test").with_db("test");
		let txn = ds.transaction(Write, Pessimistic).await?;
		txn.ensure_ns_db(None, "test", "test").await?;
		txn.commit().await?;
		execute_all(&ds, &ses, "CREATE person:tobie SET name = 'Tobie';").await?;
		// A valid GQL query parses, lowers, and executes through the
		// SurrealQL pipeline
		let mut res = ds.execute_gql("MATCH (n:person) RETURN n.name AS name", &ses, None).await?;
		assert_eq!(res.len(), 1);
		let val = res.remove(0).result?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { name: "Tobie" }]));
		// An invalid GQL query reports a parse error rather than a
		// capability error
		let err = ds.execute_gql("MATCH RETURN", &ses, None).await.unwrap_err();
		assert!(!err.to_string().contains("experimental"), "unexpected error: {err}");
		Ok(())
	}

	/// A datastore with the `gql` capability and an initialised `test/test`
	/// namespace/database, for the GQL mutation tests.
	#[cfg(feature = "gql")]
	async fn gql_test_ds() -> Result<(Datastore, Session)> {
		use crate::dbs::capabilities::Targets;
		let ds = Datastore::builder()
			.with_capabilities(Capabilities::all().with_experimental(Targets::All))
			.build_with_path("memory")
			.await?;
		let ses = Session::owner().with_ns("test").with_db("test");
		let txn = ds.transaction(Write, Pessimistic).await?;
		txn.ensure_ns_db(None, "test", "test").await?;
		txn.commit().await?;
		Ok((ds, ses))
	}

	/// Run one GQL query, asserting a single statement result, and return it.
	#[cfg(feature = "gql")]
	async fn run_gql(ds: &Datastore, ses: &Session, query: &str) -> Result<PublicValue> {
		let mut res = ds.execute_gql(query, ses, None).await?;
		assert_eq!(res.len(), 1, "expected one result for {query:?}");
		Ok(res.remove(0).result?)
	}

	/// Run one GQL query expecting failure (parse/lowering rejection or a
	/// per-statement execution error), returning the rendered error.
	#[cfg(feature = "gql")]
	async fn run_gql_err(ds: &Datastore, ses: &Session, query: &str) -> String {
		match ds.execute_gql(query, ses, None).await {
			Ok(mut res) => match res.remove(0).result {
				Ok(value) => panic!("expected {query:?} to fail, got {value:?}"),
				Err(e) => e.to_string(),
			},
			Err(e) => e.to_string(),
		}
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_set_updates_and_returns_after_image() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A', age = 30;").await?;
		// SET returns the post-mutation value.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person) WHERE n.name = 'A' SET n.age = 31 RETURN n.age AS age",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { age: 31 }]));
		// And it persisted.
		let val = run_gql(&ds, &ses, "MATCH (n:person) RETURN n.age AS age").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { age: 31 }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_remove_unsets_field_and_returns_empty() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A', age = 30;").await?;
		// A mutation-only query returns an empty array.
		let val = run_gql(&ds, &ses, "MATCH (n:person) REMOVE n.age").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		// The field is gone: a filter on the old value no longer matches.
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) WHERE n.age = 30 RETURN n.name AS name").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		// The record itself survives.
		let val = run_gql(&ds, &ses, "MATCH (n:person) RETURN n.name AS name").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { name: "A" }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_delete_nodetach_errors_with_edges_then_detach_succeeds() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(
			&ds,
			&ses,
			"CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B'; \
			 RELATE person:1->knows->person:2;",
		)
		.await?;
		// NODETACH (the default) on a node that still has edges errors.
		let err = run_gql_err(&ds, &ses, "MATCH (n:person) WHERE n.name = 'A' DELETE n").await;
		assert!(err.contains("connected edges"), "{err}");
		// The failed delete rolled back: A is still present.
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) RETURN n.name AS name ORDER BY name").await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { name: "A" },
				object! { name: "B" }
			])
		);
		// DETACH DELETE removes A and its edge.
		let val = run_gql(&ds, &ses, "MATCH (n:person) WHERE n.name = 'A' DETACH DELETE n").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) RETURN n.name AS name ORDER BY name").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { name: "B" }]));
		// The knows edge is gone too.
		let val =
			run_gql(&ds, &ses, "MATCH (a:person)-[:knows]->(b:person) RETURN a.name AS a").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_insert_node_and_edge() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		// A leading INSERT (no MATCH) creates a node exactly once.
		let val = run_gql(&ds, &ses, "INSERT (p:person {name: 'A'})").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		run_gql(&ds, &ses, "INSERT (p:person {name: 'B'})").await?;
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) RETURN n.name AS name ORDER BY name").await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { name: "A" },
				object! { name: "B" }
			])
		);
		// INSERT an edge between two MATCH-bound nodes.
		run_gql(
			&ds,
			&ses,
			"MATCH (a:person WHERE a.name = 'A') MATCH (b:person WHERE b.name = 'B') \
			 INSERT (a)-[:knows]->(b)",
		)
		.await?;
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (a:person)-[:knows]->(b:person) RETURN a.name AS a, b.name AS b",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { a: "A", b: "B" }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_label_mutation_rejected() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A';").await?;
		let err = run_gql_err(&ds, &ses, "MATCH (n:person) SET n:Archived").await;
		assert!(err.contains("Label mutation is not supported"), "{err}");
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_set_all_properties_replaces_and_multi_item() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A', age = 30, city = 'L';").await?;
		// `SET n = {…}` replaces ALL user properties (age is dropped).
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person) SET n = {name: 'A2', city: 'X'} RETURN n.name AS name, n.city AS city",
		)
		.await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![object! { name: "A2", city: "X" }])
		);
		// The dropped `age` no longer matches.
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) WHERE n.age = 30 RETURN n.name AS name").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		// Multi-item SET on distinct properties.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person) SET n.age = 5, n.city = 'Z' RETURN n.age AS age, n.city AS city",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { age: 5, city: "Z" }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_multi_statement_set_last_wins() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A', age = 1;").await?;
		// Two SET statements in one query: the second sees the first's write.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person) SET n.age = 5 SET n.age = n.age + 1 RETURN n.age AS age",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { age: 6 }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_edge_set_remove_delete() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(
			&ds,
			&ses,
			"CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B'; \
			 RELATE person:1->knows->person:2 SET since = 2020;",
		)
		.await?;
		// SET a property on a bound edge.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (a:person)-[k:knows]->(b:person) SET k.since = 2099 RETURN k.since AS since",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { since: 2099 }]));
		// DELETE a bound edge (NODETACH default is fine: an edge has no sub-edges).
		let val = run_gql(&ds, &ses, "MATCH (a:person)-[k:knows]->(b:person) DELETE k").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		let val =
			run_gql(&ds, &ses, "MATCH (a:person)-[:knows]->(b:person) RETURN a.name AS a").await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_fanout_set_is_consistent_per_row() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(
			&ds,
			&ses,
			"CREATE person:1 SET name = 'A', age = 30; CREATE person:2 SET name = 'B'; \
			 CREATE person:3 SET name = 'C'; RELATE person:1->knows->person:2; \
			 RELATE person:1->knows->person:3;",
		)
		.await?;
		// `a` fans out to two rows; SET a.age = 7 must show a consistent AFTER image
		// on both rows (the §1 fix — no stale pre-mutation binding on a duplicate).
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (a:person)-[:knows]->(b:person) WHERE a.name = 'A' SET a.age = 7 \
			 RETURN a.age AS age, b.name AS b ORDER BY b",
		)
		.await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { age: 7, b: "B" },
				object! { age: 7, b: "C" }
			])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_detach_delete_nulls_cascaded_edge_binding() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(
			&ds,
			&ses,
			"CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B'; \
			 RELATE person:1->knows->person:2 SET since = 2020;",
		)
		.await?;
		// DETACH DELETE on `a` cascades the `k` edge; the §3 fix nulls `k` in the
		// RETURN rather than surfacing the stale (deleted) edge.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (a:person)-[k:knows]->(b:person) WHERE a.name = 'A' DETACH DELETE a \
			 RETURN a AS a, k AS k, b.name AS b",
		)
		.await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![object! {
				a: PublicValue::Null,
				k: PublicValue::Null,
				b: "B"
			}])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_insert_left_edge_chain_and_repeated_anon() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B';")
			.await?;
		// A left-pointing INSERT edge relates b -> a.
		run_gql(
			&ds,
			&ses,
			"MATCH (a:person WHERE a.name = 'A') MATCH (b:person WHERE b.name = 'B') \
			 INSERT (a)<-[:knows]-(b)",
		)
		.await?;
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (x:person)-[:knows]->(y:person) RETURN x.name AS x, y.name AS y",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { x: "B", y: "A" }]));
		// A chained INSERT creates three new nodes and two edges.
		run_gql(
			&ds,
			&ses,
			"INSERT (x:item {n: 1})-[:link]->(y:item {n: 2})-[:link]->(z:item {n: 3})",
		)
		.await?;
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (a:item)-[:link]->(b:item) RETURN a.n AS a, b.n AS b ORDER BY a",
		)
		.await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { a: 1, b: 2 },
				object! { a: 2, b: 3 }
			])
		);
		// Repeated anonymous nodes in one INSERT create distinct records.
		run_gql(&ds, &ses, "INSERT (:tag {v: 1}), (:tag {v: 1})").await?;
		let val = run_gql(&ds, &ses, "MATCH (t:tag) RETURN t.v AS v ORDER BY v").await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![object! { v: 1 }, object! { v: 1 }])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_mutation_rejection_ledger() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		// Each lowers (or parses) to a precise rejection. No data needed — these
		// fail before execution.
		let cases: &[(&str, &str)] = &[
			// Mutating a variable not bound by the read body.
			("MATCH (n:person) SET x.age = 1", "Unknown variable"),
			// Mutating a path variable (a composite, not a record).
			("MATCH p = (a:person)-[:knows]->(b:person) SET p.x = 1", "group or path variable"),
			// `SET a = {…}` setting a reserved key.
			("MATCH (n:person) SET n = {id: 1}", "reserved `id` key"),
			// Per-property SET of a reserved edge endpoint (the native write path
			// would otherwise silently re-stamp `out`, so it is rejected up front).
			("MATCH (a:person)-[k:knows]->(b:person) SET k.out = 1", "reserved `out` key"),
			// Label mutation via REMOVE.
			("MATCH (n:person) REMOVE n:Foo", "Label mutation is not supported"),
			// INSERT re-declaring a MATCH-bound variable as a new (labelled) node.
			("MATCH (a:person) INSERT (a:thing)", "already bound"),
			// INSERT node that is neither labelled nor a bound-variable reference.
			("INSERT (x)", "must declare a label"),
			// Undirected INSERT edge.
			("INSERT (a:person)~[:knows]~(b:person)", "Undirected INSERT edges"),
			// DELETE of a non-variable expression.
			("MATCH (n:person) DELETE n.age", "bound variable"),
		];
		for (query, expected) in cases {
			let err = run_gql_err(&ds, &ses, query).await;
			assert!(err.contains(expected), "query {query:?}: expected {expected:?}, got: {err}");
		}
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_mutation_respects_record_permissions() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		// Owner: a table that denies create/update (but allows select), seeded
		// with one row.
		execute_all(
			&ds,
			&ses,
			"DEFINE TABLE person SCHEMALESS \
			 PERMISSIONS FOR select FULL, FOR create NONE, FOR update NONE, FOR delete NONE; \
			 CREATE person:1 SET name = 'A', age = 30;",
		)
		.await?;
		// A record-scoped session: table PERMISSIONS clauses apply (a role-based
		// system session would bypass them, so this is the meaningful test).
		let rec = Session::for_record(
			"test",
			"test",
			"user",
			surrealdb_types::Value::RecordId(surrealdb_types::RecordId::new("user", "tester")),
		);
		// Each write goes through the native document pipeline, so each is denied:
		// INSERT (create), SET (update), DELETE (delete) all leave the row intact.
		let _ = ds.execute_gql("INSERT (p:person {name: 'B'})", &rec, None).await;
		let _ = ds.execute_gql("MATCH (n:person) SET n.age = 99", &rec, None).await;
		let _ = ds.execute_gql("MATCH (n:person) DETACH DELETE n", &rec, None).await;
		// As owner: exactly the original row remains, unchanged.
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) RETURN n.name AS name, n.age AS age").await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![object! { name: "A", age: 30 }])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_nodetach_probe_ignores_select_permissions() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		// `person` is fully visible/deletable to a record session, but the `knows`
		// edge table hides SELECT from it. Seed a node with one connected edge.
		execute_all(
			&ds,
			&ses,
			"DEFINE TABLE person SCHEMALESS \
			 PERMISSIONS FOR select FULL, FOR create FULL, FOR update FULL, FOR delete FULL; \
			 DEFINE TABLE knows SCHEMALESS \
			 PERMISSIONS FOR select NONE, FOR create FULL, FOR update FULL, FOR delete FULL; \
			 CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B'; \
			 RELATE person:1->knows->person:2;",
		)
		.await?;
		let rec = Session::for_record(
			"test",
			"test",
			"user",
			surrealdb_types::Value::RecordId(surrealdb_types::RecordId::new("user", "tester")),
		);
		// The NODETACH (default) connected-edge probe runs with permissions
		// DISABLED, so it sees the `knows` edge the record session cannot SELECT and
		// errors — rather than passing the guard and letting the native DELETE
		// silently cascade an edge the caller opted out of removing.
		let err = run_gql_err(&ds, &rec, "MATCH (n:person WHERE n.name = 'A') DELETE n").await;
		assert!(err.contains("connected edges"), "{err}");
		// The node (and its edge) survive the rejected delete.
		let val =
			run_gql(&ds, &ses, "MATCH (n:person) RETURN n.name AS name ORDER BY name").await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { name: "A" },
				object! { name: "B" }
			])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_match_after_set_rereads_live_state() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(
			&ds,
			&ses,
			"CREATE person:1 SET name = 'A', tier = 'bronze'; \
			 CREATE person:2 SET name = 'B', tier = 'gold';",
		)
		.await?;
		// A read step after a write re-scans the live (post-write) state in the same
		// transaction: promoting A to gold and then MATCHing on `tier = 'gold'` finds
		// both A (just written) and B.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person WHERE n.name = 'A') SET n.tier = 'gold' \
			 MATCH (m:person WHERE m.tier = 'gold') RETURN m.name AS name ORDER BY name",
		)
		.await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { name: "A" },
				object! { name: "B" }
			])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_match_after_delete_rereads_live_state() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B';")
			.await?;
		// The trailing MATCH no longer observes the row the DELETE removed.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person WHERE n.name = 'A') DETACH DELETE n \
			 MATCH (m:person) RETURN m.name AS name ORDER BY name",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { name: "B" }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_match_after_insert_sees_new_node_and_anchors_on_binding() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A', age = 20;").await?;
		// A trailing MATCH re-scans and observes the just-inserted node, and can join
		// on the binding the INSERT created (`a.age` anchors the predicate on `b`).
		let val = run_gql(
			&ds,
			&ses,
			"INSERT (a:person {name: 'New', age: 20}) \
			 MATCH (b:person WHERE b.age = a.age) RETURN b.name AS name ORDER BY name",
		)
		.await?;
		assert_eq!(
			val,
			PublicValue::Array(surrealdb_types::array![
				object! { name: "A" },
				object! { name: "New" }
			])
		);
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_optional_match_after_set_rereads_live_state() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A'; CREATE person:2 SET name = 'B';")
			.await?;
		// An OPTIONAL block after a write takes the general left-join path (the
		// mutating accumulator is the join's probe side, the OPTIONAL read is its
		// build side). The OPTIONAL must still observe the write: after promoting A
		// to gold, `OPTIONAL MATCH (m:person WHERE m.tier = 'gold')` matches A (just
		// written), so `m` is bound, NOT null-filled.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (n:person WHERE n.name = 'A') SET n.tier = 'gold' \
			 OPTIONAL MATCH (m:person WHERE m.tier = 'gold') \
			 RETURN n.name AS n, m.name AS m",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { n: "A", m: "A" }]));
		Ok(())
	}

	#[cfg(feature = "gql")]
	#[tokio::test]
	async fn gql_optional_match_after_insert_sees_new_node() -> Result<()> {
		let (ds, ses) = gql_test_ds().await?;
		execute_all(&ds, &ses, "CREATE person:1 SET name = 'A';").await?;
		// An OPTIONAL block after an INSERT must see the just-created node. The
		// INSERT runs once per matched `person`; the trailing OPTIONAL re-scans
		// `tag` and binds `m` to the new node rather than null-filling it.
		let val = run_gql(
			&ds,
			&ses,
			"MATCH (a:person WHERE a.name = 'A') INSERT (t:tag {name: 'X'}) \
			 OPTIONAL MATCH (m:tag) \
			 RETURN m.name AS m",
		)
		.await?;
		assert_eq!(val, PublicValue::Array(surrealdb_types::array![object! { m: "X" }]));
		Ok(())
	}

	async fn index_key_base(ds: &Datastore, table: &str, index: &str) -> Result<IndexKeyBase> {
		let txn = ds.transaction(Read, Optimistic).await?;
		let ns = txn.get_ns_by_name("test", None).await?.unwrap();
		let db = txn.get_db_by_name("test", "test", None).await?.unwrap();
		let table = TableName::from(table);
		let ix =
			txn.get_tb_index(ns.namespace_id, db.database_id, &table, index, None).await?.unwrap();
		txn.cancel().await?;
		Ok(IndexKeyBase::new(ns.namespace_id, db.database_id, table, ix.index_id))
	}

	async fn assert_index_compaction_commit_retry(
		site: RetryableConflictSite,
		table: &str,
		index: &str,
		sql: &str,
	) -> Result<()> {
		let (ds, session) = new_index_compaction_test_ds().await?;
		execute_all(&ds, &session, sql).await?;
		let ikb = index_key_base(&ds, table, index).await?;
		let node_id = ds.id();
		let _guard = inject_retryable_conflict(site, node_id);

		ds.process_index_compaction(&ikb, CancellationToken::new()).await?;

		assert_eq!(retryable_conflict_count(site, node_id), 0);
		Ok(())
	}

	const COUNT_COMPACTION_SQL: &str = "
		DEFINE TABLE user SCHEMALESS;
		DEFINE INDEX count_idx ON user COUNT;
		CREATE user:1 SET name = 'one' RETURN NONE;
		CREATE user:2 SET name = 'two' RETURN NONE;
	";

	const FULLTEXT_COMPACTION_SQL: &str = "
		DEFINE ANALYZER simple TOKENIZERS blank FILTERS lowercase;
		DEFINE TABLE doc SCHEMALESS;
		DEFINE INDEX ft_idx ON doc FIELDS text FULLTEXT ANALYZER simple BM25 HIGHLIGHTS;
		CREATE doc:1 SET text = 'alpha beta' RETURN NONE;
		CREATE doc:2 SET text = 'beta gamma' RETURN NONE;
	";

	const HNSW_COMPACTION_SQL: &str = "
		DEFINE TABLE vec SCHEMALESS;
		DEFINE INDEX hnsw_idx ON vec FIELDS vector HNSW DIMENSION 2 DIST EUCLIDEAN TYPE F32 EFC 16 M 4;
		CREATE vec:1 SET vector = [1, 2] RETURN NONE;
		CREATE vec:2 SET vector = [2, 3] RETURN NONE;
	";

	#[tokio::test]
	async fn count_index_compaction_retries_commit_conflict() -> Result<()> {
		assert_index_compaction_commit_retry(
			RetryableConflictSite::CountCompaction,
			"user",
			"count_idx",
			COUNT_COMPACTION_SQL,
		)
		.await
	}

	#[tokio::test]
	async fn fulltext_index_compaction_retries_commit_conflict() -> Result<()> {
		assert_index_compaction_commit_retry(
			RetryableConflictSite::FullTextCompaction,
			"doc",
			"ft_idx",
			FULLTEXT_COMPACTION_SQL,
		)
		.await
	}

	#[tokio::test]
	async fn hnsw_index_compaction_retries_commit_conflict() -> Result<()> {
		assert_index_compaction_commit_retry(
			RetryableConflictSite::HnswCompaction,
			"vec",
			"hnsw_idx",
			HNSW_COMPACTION_SQL,
		)
		.await
	}

	#[tokio::test]
	async fn index_compaction_retries_queue_cleanup_commit_conflict() -> Result<()> {
		let (ds, session) = new_index_compaction_test_ds().await?;
		execute_all(&ds, &session, COUNT_COMPACTION_SQL).await?;
		let site = RetryableConflictSite::IndexCompactionQueueCleanup;
		let node_id = ds.id();
		let _guard = inject_retryable_conflict(site, node_id);

		let (_, errors) = Datastore::index_compaction(
			Arc::new(ds),
			Duration::from_secs(1),
			CancellationToken::new(),
		)
		.await?;

		assert_eq!(errors, 0);
		assert_eq!(retryable_conflict_count(site, node_id), 0);
		Ok(())
	}

	#[tokio::test]
	async fn archive_node_for_shutdown_reports_success() {
		let outcome = archive_node_for_shutdown(Duration::from_secs(60), Ok(()));

		assert_eq!(outcome, ShutdownNodeDeleteOutcome::Archived);
	}

	#[tokio::test]
	async fn archive_node_for_shutdown_reports_failure() {
		let outcome = archive_node_for_shutdown(
			Duration::from_secs(60),
			Err(anyhow::anyhow!("delete failed")),
		);

		assert_eq!(outcome, ShutdownNodeDeleteOutcome::Failed);
	}

	#[tokio::test]
	async fn archive_node_for_shutdown_reports_timeout() {
		let outcome = archive_node_for_shutdown(
			Duration::from_millis(1),
			Err(anyhow::Error::new(Error::QueryTimedout(Duration::from_millis(1).into()))),
		);

		assert_eq!(outcome, ShutdownNodeDeleteOutcome::TimedOut);
	}

	#[tokio::test]
	async fn node_tx_step_cancels_after_timeout() {
		let ds = Datastore::new("memory").await.unwrap();
		let txn = ds.transaction(Write, Optimistic).await.unwrap();
		let timeout_duration = Duration::from_millis(10);

		let err = await_node_tx_step(
			&txn,
			Instant::now() + timeout_duration,
			timeout_duration,
			None,
			pending::<Result<()>>(),
		)
		.await
		.unwrap_err();

		assert!(matches!(err.downcast_ref::<Error>(), Some(Error::QueryTimedout(_))));
		assert!(txn.closed());
	}

	#[tokio::test]
	async fn node_tx_step_cancels_after_cancellation() {
		let ds = Datastore::new("memory").await.unwrap();
		let txn = ds.transaction(Write, Optimistic).await.unwrap();
		let canceller = CancellationToken::new();
		canceller.cancel();

		let err = await_node_tx_step(
			&txn,
			Instant::now() + Duration::from_secs(60),
			Duration::from_secs(60),
			Some(&canceller),
			pending::<Result<()>>(),
		)
		.await
		.unwrap_err();

		assert!(matches!(err.downcast_ref::<Error>(), Some(Error::QueryCancelled)));
		assert!(txn.closed());
	}

	#[tokio::test]
	async fn node_tx_step_cancels_after_error() {
		let ds = Datastore::new("memory").await.unwrap();
		let txn = ds.transaction(Write, Optimistic).await.unwrap();

		let err = await_node_tx_step(
			&txn,
			Instant::now() + Duration::from_secs(60),
			Duration::from_secs(60),
			None,
			async { Err::<(), _>(anyhow::anyhow!("step failed")) },
		)
		.await
		.unwrap_err();

		assert_eq!(err.to_string(), "step failed");
		assert!(txn.closed());
	}

	#[tokio::test]
	async fn node_tx_step_success_leaves_transaction_open() {
		let ds = Datastore::new("memory").await.unwrap();
		let txn = ds.transaction(Write, Optimistic).await.unwrap();

		await_node_tx_step(
			&txn,
			Instant::now() + Duration::from_secs(60),
			Duration::from_secs(60),
			None,
			async { Ok::<_, anyhow::Error>(()) },
		)
		.await
		.unwrap();

		assert!(!txn.closed());
		txn.commit().await.unwrap();
		assert!(txn.closed());
	}

	#[tokio::test]
	async fn node_heartbeat_age_is_small_after_insert() {
		let ds = Datastore::new("memory").await.unwrap();
		// `insert_node` writes the current node's heartbeat at `clock_now()`.
		ds.insert_node().await.unwrap();
		let age = ds.node_heartbeat_age().await.unwrap();
		assert!(age < Duration::from_secs(5), "heartbeat should be fresh, got {age:?}");
	}

	#[tokio::test]
	async fn node_heartbeat_age_reflects_a_stale_heartbeat() {
		let ds = Datastore::new("memory").await.unwrap();
		// Write this node's registration with a heartbeat 60s in the past.
		let now = ds.clock_now().value;
		let stale = Node::new(
			ds.id(),
			Timestamp {
				value: now.saturating_sub(60_000),
			},
			false,
		);
		let key = crate::key::root::nd::new(ds.id());
		let txn = ds.transaction(Write, Optimistic).await.unwrap();
		txn.set(&key, &stale).await.unwrap();
		txn.commit().await.unwrap();
		// The reported age should reflect the stale heartbeat.
		let age = ds.node_heartbeat_age().await.unwrap();
		assert!(age >= Duration::from_secs(59), "heartbeat should be stale, got {age:?}");
	}

	#[tokio::test]
	async fn test_setup_superuser() {
		let ds = Datastore::new("memory").await.unwrap();
		let username = "root";
		let password = "root";

		// Setup the initial user if there are no root users
		{
			let txn = ds.transaction(Read, Optimistic).await.unwrap();
			assert_eq!(txn.all_root_users(None).await.unwrap().len(), 0);
			txn.cancel().await.unwrap();
		}
		ds.initialise_credentials(username, password).await.unwrap();
		{
			let txn = ds.transaction(Read, Optimistic).await.unwrap();
			assert_eq!(txn.all_root_users(None).await.unwrap().len(), 1);
			txn.cancel().await.unwrap();
		}
		verify_root_creds(&ds, username, password).await.unwrap();

		// Do not setup the initial root user if there are root users:
		// Test the scenario by making sure the custom password doesn't change.
		let sql = "DEFINE USER root ON ROOT PASSWORD 'test' ROLES OWNER";
		let sess = Session::owner();
		ds.execute(sql, &sess, None).await.unwrap();
		let pass_hash = {
			let txn = ds.transaction(Read, Optimistic).await.unwrap();
			let res = txn.expect_root_user(username).await.unwrap().hash.clone();
			txn.cancel().await.unwrap();
			res
		};

		ds.initialise_credentials(username, password).await.unwrap();
		{
			let txn = ds.transaction(Read, Optimistic).await.unwrap();
			assert_eq!(pass_hash, txn.expect_root_user(username).await.unwrap().hash.clone());
			txn.cancel().await.unwrap();
		}
	}

	#[tokio::test]
	pub async fn very_deep_query() -> Result<()> {
		use reblessive::{Stack, Stk};

		use crate::expr::{BinaryOperator, Expr, Literal};
		use crate::kvs::Datastore;
		use crate::val::{Number, Value};

		// build query manually to bypass query limits.
		let mut stack = Stack::new();
		async fn build_query(stk: &mut Stk, depth: usize) -> Expr {
			if depth == 0 {
				Expr::Binary {
					left: Box::new(Expr::Literal(Literal::Integer(1))),
					op: BinaryOperator::Add,
					right: Box::new(Expr::Literal(Literal::Integer(1))),
				}
			} else {
				let q = stk.run(|stk| build_query(stk, depth - 1)).await;
				Expr::Binary {
					left: Box::new(q),
					op: BinaryOperator::Add,
					right: Box::new(Expr::Literal(Literal::Integer(1))),
				}
			}
		}
		let val = stack.enter(|stk| build_query(stk, 1000)).finish();

		let dbs = Datastore::builder()
			.with_capabilities(Capabilities::all())
			.build_with_path("memory")
			.await
			.unwrap();

		let opt = Options::new(&dbs.config())
			.with_ns(Some("test".into()))
			.with_db(Some("test".into()))
			.with_max_computation_depth(u32::MAX);

		// Create a default context
		let mut ctx = dbs.setup_ctx()?;
		// Start a new transaction
		let txn = dbs.transaction(TransactionType::Read, Optimistic).await?.enclose();
		// Store the transaction
		ctx.set_transaction(Arc::clone(&txn));
		// Freeze the context
		let ctx = ctx.freeze();
		// Compute the value
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack
			.enter(|stk| val.compute(stk, &ctx, &opt, None))
			.finish()
			.await
			.catch_return()
			.unwrap();
		assert_eq!(res, Value::Number(Number::Int(1002)));
		txn.cancel().await?;
		Ok(())
	}

	#[tokio::test]
	async fn cross_transaction_caching_uuids_updated() -> Result<()> {
		let (send, _recv) = crate::channel::bounded(crate::cnf::NOTIFICATIONS_CHANNEL_SIZE);
		let ds = Datastore::builder()
			.with_capabilities(Capabilities::all())
			.with_notify(send)
			.build_with_path("memory")
			.await?;
		let ses = Session::owner().with_ns("test").with_db("test").with_rt(true);

		let db = {
			let txn = ds.transaction(TransactionType::Write, LockType::Pessimistic).await?;
			let db = txn.ensure_ns_db(None, "test", "test").await?;
			txn.commit().await?;
			db
		};

		// Define the table, set the initial uuids
		let initial = {
			let sql = r"DEFINE TABLE test;".to_owned();
			let res = &mut ds.execute(&sql, &ses, None).await?;
			assert_eq!(res.len(), 1);
			res.remove(0).result.unwrap();
			// Obtain the initial uuids
			let txn = ds.transaction(TransactionType::Read, LockType::Pessimistic).await?;
			let tb = TableName::from("test");
			let initial = txn.get_tb(db.namespace_id, db.database_id, &tb, None).await?.unwrap();
			txn.cancel().await?;
			initial
		};

		// Define some resources to refresh the UUIDs
		let lqid = {
			let sql = r"
		DEFINE FIELD test ON test;
		DEFINE EVENT test ON test WHEN {} THEN {};
		DEFINE TABLE view AS SELECT * FROM test;
		DEFINE INDEX test ON test FIELDS test;
		LIVE SELECT * FROM test;
	"
			.to_owned();
			let res = &mut ds.execute(&sql, &ses, None).await?;
			assert_eq!(res.len(), 5);
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
			let lqid = res.remove(0).result?;
			assert!(matches!(lqid, PublicValue::Uuid(_)));
			lqid
		};

		// Obtain the uuids after definitions
		let after_define = {
			let txn = ds.transaction(TransactionType::Read, LockType::Pessimistic).await?;
			let tb = TableName::from("test");
			let after_define =
				txn.get_tb(db.namespace_id, db.database_id, &tb, None).await?.unwrap();
			txn.cancel().await?;
			// Compare uuids after definitions
			assert_ne!(initial.cache_indexes_ts, after_define.cache_indexes_ts);
			assert_ne!(initial.cache_tables_ts, after_define.cache_tables_ts);
			assert_ne!(initial.cache_events_ts, after_define.cache_events_ts);
			assert_ne!(initial.cache_fields_ts, after_define.cache_fields_ts);
			// LIVE bumped the committed live-query cache timestamp.
			assert_ne!(initial.cache_lives_ts, after_define.cache_lives_ts);
			after_define
		};

		// Remove the defined resources to refresh the UUIDs
		{
			let sql = r"
		REMOVE FIELD test ON test;
		REMOVE EVENT test ON test;
		REMOVE TABLE view;
		REMOVE INDEX test ON test;
		KILL $lqid;
	"
			.to_owned();
			let vars =
				PublicVariables::from(BTreeMap::from_iter(map! { "lqid".to_string() => lqid }));
			let res = &mut ds.execute(&sql, &ses, Some(vars)).await?;
			assert_eq!(res.len(), 5);
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
			res.remove(0).result.unwrap();
		}
		// Obtain the uuids after definitions
		{
			let txn = ds.transaction(TransactionType::Read, LockType::Pessimistic).await?;
			let tb = TableName::from("test");
			let after_remove =
				txn.get_tb(db.namespace_id, db.database_id, &tb, None).await?.unwrap();
			txn.cancel().await?;
			// Compare uuids after definitions
			assert_ne!(after_define.cache_fields_ts, after_remove.cache_fields_ts);
			assert_ne!(after_define.cache_events_ts, after_remove.cache_events_ts);
			assert_ne!(after_define.cache_tables_ts, after_remove.cache_tables_ts);
			assert_ne!(after_define.cache_indexes_ts, after_remove.cache_indexes_ts);
			// KILL bumped the committed live-query cache timestamp.
			assert_ne!(after_define.cache_lives_ts, after_remove.cache_lives_ts);
		}
		//
		Ok(())
	}
}