surrealdb-server 3.3.1

use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use clap::Args;
use rand::Rng;
use surrealdb::opt::capabilities::Capabilities as SdkCapabilities;
use surrealdb_cnf::ConfigMap;
use surrealdb_core::channel::Receiver;
use surrealdb_core::kvs::{Datastore, TransactionBuilderFactory};
use surrealdb_core::options::EngineOptions;
use surrealdb_observe::ExecutionObserver;
use surrealdb_rpc::capabilities::{
	ArbitraryQueryTarget, Capabilities, EvalQueryTarget, ExperimentalTarget, FuncTarget,
	MethodTarget, NetTarget, NewPlannerStrategy, RouteTarget, Targets,
};
use surrealdb_types::Notification;
use tokio::time::{Instant, sleep, timeout};
use tokio_util::sync::CancellationToken;

use crate::cli::Config;
use crate::core::dbs::Session;

const TARGET: &str = "surreal::dbs";

/// How often a startup operation that is still retrying reports that it is
/// still going.
///
/// The cadence is on the log, not on the retries: attempts keep to their own
/// back-off, and one failure in every interval is reported at `info` so the
/// wait is visible without `debug` logging turned on.
const STARTUP_RETRY_LOG_INTERVAL: Duration = Duration::from_secs(5);

#[derive(Args, Debug)]
pub struct StartCommandDbsOptions {
	#[arg(help = "Whether strict mode is enabled on this database instance")]
	#[arg(env = "SURREAL_STRICT", short = 's', long = "strict", hide = true)]
	strict_mode: Option<bool>,
	#[arg(help = "The maximum duration that a set of statements can run for")]
	#[arg(env = "SURREAL_QUERY_TIMEOUT", long)]
	#[arg(value_parser = super::cli::validator::duration)]
	query_timeout: Option<Duration>,
	#[arg(help = "The maximum duration that any single transaction can run for")]
	#[arg(env = "SURREAL_TRANSACTION_TIMEOUT", long)]
	#[arg(value_parser = super::cli::validator::duration)]
	transaction_timeout: Option<Duration>,
	#[arg(
		help = "The maximum duration that each built-in startup datastore operation can retry for"
	)]
	#[arg(env = "SURREAL_STARTUP_OPERATION_TIMEOUT", long, default_value = "60s")]
	#[arg(value_parser = super::cli::validator::duration)]
	startup_operation_timeout: Duration,
	#[arg(help = "Whether to allow unauthenticated access", help_heading = "Authentication")]
	#[arg(env = "SURREAL_UNAUTHENTICATED", long = "unauthenticated")]
	#[arg(default_value_t = false)]
	unauthenticated: bool,
	#[command(flatten)]
	#[command(next_help_heading = "Capabilities")]
	capabilities: DbsCapabilities,
	#[arg(help = "Sets the directory for storing temporary database files")]
	#[arg(env = "SURREAL_TEMPORARY_DIRECTORY", long = "temporary-directory")]
	#[arg(value_parser = super::cli::validator::dir_exists)]
	temporary_directory: Option<PathBuf>,
	#[arg(help = "Path to a SurrealQL file that will be imported when starting the server")]
	#[arg(env = "SURREAL_IMPORT_FILE", long = "import-file")]
	#[arg(value_parser = super::cli::validator::file_exists)]
	import_file: Option<PathBuf>,
	// Slow query logging configuration. When `slow_log_threshold` is set, any
	// statement taking longer than the threshold will be logged along with a
	// normalized, single-line SQL rendering. You can control which `$param`
	// values appear in the log with the following lists:
	// - `slow_log_param_deny` takes precedence and excludes matches.
	// - If `slow_log_param_allow` is non-empty, only listed parameter names are included;
	//   otherwise all parameters are allowed by default (subject to deny).
	#[arg(help = "The minimum execution time in milliseconds to trigger slow query logging")]
	#[arg(env = "SURREAL_SLOW_QUERY_LOG_THRESHOLD", long = "slow-log-threshold")]
	#[arg(value_parser = super::cli::validator::duration)]
	slow_log_threshold: Option<Duration>,
	#[arg(help = "A comma-separated list of parameter names to include in slow query logs")]
	#[arg(env = "SURREAL_SLOW_QUERY_LOG_PARAM_ALLOW", long = "slow-log-param-allow")]
	#[arg(value_delimiter = ',', num_args = 1..)]
	slow_log_param_allow: Vec<String>,
	#[arg(help = "A comma-separated list of parameter names to omit from slow query logs")]
	#[arg(env = "SURREAL_SLOW_QUERY_LOG_PARAM_DENY", long = "slow-log-param-deny")]
	#[arg(value_delimiter = ',', num_args = 1..)]
	slow_log_param_deny: Vec<String>,
	#[arg(help = "The default namespace for a new instance")]
	#[arg(env = "SURREAL_DEFAULT_NAMESPACE", long = "default-namespace")]
	default_namespace: Option<String>,
	#[arg(help = "The default database for a new instance")]
	#[arg(env = "SURREAL_DEFAULT_DATABASE", long = "default-database")]
	default_database: Option<String>,
	#[arg(help = "Whether to disable default namespace and database creation")]
	#[arg(env = "SURREAL_NO_DEFAULTS", long = "no-defaults", conflicts_with_all = ["default_namespace", "default_database"])]
	#[arg(default_value_t = false)]
	no_defaults: bool,
	#[arg(help = "Load Surrealism modules lazily on first use instead of eagerly at startup")]
	#[arg(env = "SURREAL_LAZY_SURREALISM", long = "lazy-surrealism")]
	#[arg(default_value_t = false)]
	lazy_surrealism: bool,
}

#[derive(Args, Debug)]
pub struct DbsCapabilities {
	//
	// Allow
	#[arg(help = "Allow all capabilities except for those more specifically denied")]
	#[arg(env = "SURREAL_CAPS_ALLOW_ALL", short = 'A', long, conflicts_with = "deny_all")]
	allow_all: bool,

	#[cfg(feature = "scripting")]
	#[arg(help = "Allow execution of embedded scripting functions")]
	#[arg(env = "SURREAL_CAPS_ALLOW_SCRIPT", long, conflicts_with_all = ["allow_all", "deny_scripting"])]
	allow_scripting: bool,

	#[arg(help = "Allow guest users to execute queries")]
	#[arg(env = "SURREAL_CAPS_ALLOW_GUESTS", long, conflicts_with_all = ["allow_all", "deny_guests"])]
	allow_guests: bool,

	#[arg(
		help = "Allow execution of all functions except for functions that are specifically denied. Alternatively, you can provide a comma-separated list of function names to allow",
		long_help = r#"Allow execution of all functions except for functions that are specifically denied. Alternatively, you can provide a comma-separated list of function names to allow
Specifically denied functions and function families prevail over any other allowed function execution.
Function names must be in the form <family>[::<name>]. For example:
 - 'http' or 'http::*' -> Include all functions in the 'http' family
 - 'http::get' -> Include only the 'get' function in the 'http' family
"#
	)]
	#[arg(env = "SURREAL_CAPS_ALLOW_FUNC", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::func_targets)]
	allow_funcs: Option<Targets<FuncTarget>>,

	#[arg(hide = true)]
	#[arg(env = "SURREAL_CAPS_ALLOW_EXPERIMENTAL", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::experimental_targets)]
	allow_experimental: Option<Targets<ExperimentalTarget>>,

	#[arg(
		help = "Allow execution of arbitrary queries by certain user groups except when specifically denied.",
		long_help = r#"Allow execution of arbitrary queries by certain user groups except when specifically denied. Alternatively, you can provide a comma-separated list of user groups to allow
Specifically denied user groups prevail over any other allowed user group.
User groups must be one of "guest", "record" or "system".
"#
	)]
	#[arg(env = "SURREAL_CAPS_ALLOW_ARBITRARY_QUERY", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::query_arbitrary_targets)]
	allow_arbitrary_query: Option<Targets<ArbitraryQueryTarget>>,

	#[arg(
		help = "Allow the eval::* functions to be invoked by certain user groups. Denied for everyone by default.",
		long_help = r#"Allow the eval::surql / eval::gql functions to be invoked by certain user groups. Provide a comma-separated list of user groups to allow. This is an additive gate on top of arbitrary-query: an eval invocation must also satisfy the arbitrary-query capability.
Specifically denied user groups prevail over any other allowed user group.
User groups must be one of "guest", "record" or "system".
Denied for everyone by default (even under --allow-all); must be explicitly enabled.
"#
	)]
	#[arg(env = "SURREAL_CAPS_ALLOW_EVAL_QUERY", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::eval_query_targets)]
	allow_eval_query: Option<Targets<EvalQueryTarget>>,

	#[arg(
		help = "Allow all outbound network connections except for network targets that are specifically denied. Alternatively, you can provide a comma-separated list of network targets to allow",
		long_help = r#"Allow all outbound network connections except for network targets that are specifically denied. Alternatively, you can provide a comma-separated list of network targets to allow
Specifically denied network targets prevail over any other allowed outbound network connections.
Targets must be in the form of <host>[:<port>], <ipv4|ipv6>[/<mask>]. For example:
 - 'surrealdb.com', '127.0.0.1' or 'fd00::1' -> Match outbound connections to these hosts on any port
 - 'surrealdb.com:80', '127.0.0.1:80' or 'fd00::1:80' -> Match outbound connections to these hosts on port 80
 - '10.0.0.0/8' or 'fd00::/8' -> Match outbound connections to any host in these networks
"#
	)]
	#[arg(env = "SURREAL_CAPS_ALLOW_NET", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::net_targets)]
	allow_net: Option<Targets<NetTarget>>,

	#[arg(
		help = "Allow all RPC methods to be called except for routes that are specifically denied. Alternatively, you can provide a comma-separated list of RPC methods to allow."
	)]
	#[arg(env = "SURREAL_CAPS_ALLOW_RPC", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(default_value_os = "")] // Allow all RPC methods by default
	#[arg(value_parser = super::cli::validator::method_targets)]
	allow_rpc: Option<Targets<MethodTarget>>,

	#[arg(
		help = "Allow all HTTP routes to be requested except for routes that are specifically denied. Alternatively, you can provide a comma-separated list of HTTP routes to allow."
	)]
	#[arg(env = "SURREAL_CAPS_ALLOW_HTTP", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(default_value_os = "")] // Allow all HTTP routes by default
	#[arg(value_parser = super::cli::validator::route_targets)]
	allow_http: Option<Targets<RouteTarget>>,

	//
	// Deny
	#[arg(help = "Deny all capabilities except for those more specifically allowed")]
	#[arg(env = "SURREAL_CAPS_DENY_ALL", short = 'D', long, conflicts_with = "allow_all")]
	deny_all: bool,

	#[cfg(feature = "scripting")]
	#[arg(help = "Deny execution of embedded scripting functions")]
	#[arg(env = "SURREAL_CAPS_DENY_SCRIPT", long, conflicts_with_all = ["deny_all", "allow_scripting"])]
	deny_scripting: bool,

	#[arg(help = "Deny guest users to execute queries")]
	#[arg(env = "SURREAL_CAPS_DENY_GUESTS", long, conflicts_with_all = ["deny_all", "allow_guests"])]
	deny_guests: bool,

	#[arg(
		help = "Deny execution of all functions except for functions that are specifically allowed. Alternatively, you can provide a comma-separated list of function names to deny",
		long_help = r#"Deny execution of all functions except for functions that are specifically allowed. Alternatively, you can provide a comma-separated list of function names to deny.
Specifically allowed functions and function families prevail over a general denial of function execution.
Function names must be in the form <family>[::<name>]. For example:
 - 'http' or 'http::*' -> Include all functions in the 'http' family
 - 'http::get' -> Include only the 'get' function in the 'http' family
"#
	)]
	#[arg(env = "SURREAL_CAPS_DENY_FUNC", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::func_targets)]
	deny_funcs: Option<Targets<FuncTarget>>,

	#[arg(hide = true)]
	#[arg(env = "SURREAL_CAPS_DENY_EXPERIMENTAL", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::experimental_targets)]
	deny_experimental: Option<Targets<ExperimentalTarget>>,

	#[arg(
		help = "Deny execution of arbitrary queries by certain user groups except when specifically allowed.",
		long_help = r#"Deny execution of arbitrary queries by certain user groups except when specifically allowed. Alternatively, you can provide a comma-separated list of user groups to deny
Specifically allowed user groups prevail over a general denial of user group.
User groups must be one of "guest", "record" or "system".
"#
	)]
	#[arg(env = "SURREAL_CAPS_DENY_ARBITRARY_QUERY", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::query_arbitrary_targets)]
	deny_arbitrary_query: Option<Targets<ArbitraryQueryTarget>>,

	#[arg(
		help = "Deny the eval::* functions for certain user groups except when specifically allowed.",
		long_help = r#"Deny the eval::surql / eval::gql functions for certain user groups. Provide a comma-separated list of user groups to deny.
Specifically denied user groups prevail over any allowed user group.
User groups must be one of "guest", "record" or "system".
"#
	)]
	#[arg(env = "SURREAL_CAPS_DENY_EVAL_QUERY", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::eval_query_targets)]
	deny_eval_query: Option<Targets<EvalQueryTarget>>,

	#[arg(
		help = "Deny all outbound network connections except for network targets that are specifically allowed. Alternatively, you can provide a comma-separated list of network targets to deny",
		long_help = r#"Deny all outbound network connections except for network targets that are specifically allowed. Alternatively, you can provide a comma-separated list of network targets to deny.
Specifically allowed network targets prevail over a general denial of outbound network connections.
Targets must be in the form of <host>[:<port>], <ipv4|ipv6>[/<mask>]. For example:
 - 'surrealdb.com', '127.0.0.1' or 'fd00::1' -> Match outbound connections to these hosts on any port
 - 'surrealdb.com:80', '127.0.0.1:80' or 'fd00::1:80' -> Match outbound connections to these hosts on port 80
 - '10.0.0.0/8' or 'fd00::/8' -> Match outbound connections to any host in these networks
"#
	)]
	#[arg(env = "SURREAL_CAPS_DENY_NET", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::net_targets)]
	deny_net: Option<Targets<NetTarget>>,

	#[arg(
		help = "Deny all RPC methods from being called except for methods that are specifically allowed. Alternatively, you can provide a comma-separated list of RPC methods to deny."
	)]
	#[arg(env = "SURREAL_CAPS_DENY_RPC", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::method_targets)]
	deny_rpc: Option<Targets<MethodTarget>>,

	#[arg(
		help = "Deny all HTTP routes from being requested except for routes that are specifically allowed. Alternatively, you can provide a comma-separated list of HTTP routes to deny."
	)]
	#[arg(env = "SURREAL_CAPS_DENY_HTTP", long)]
	// If the arg is provided without value, then assume it's "", which gets parsed into
	// Targets::All
	#[arg(default_missing_value_os = "", num_args = 0..)]
	#[arg(value_parser = super::cli::validator::route_targets)]
	deny_http: Option<Targets<RouteTarget>>,

	#[arg(
		help = "Strategy for the streaming query planner: 'best-effort' (default), 'compute-only', or 'all-read-only'"
	)]
	#[arg(env = "SURREAL_PLANNER_STRATEGY", long = "planner-strategy")]
	#[arg(default_value = "best-effort")]
	planner_strategy: NewPlannerStrategy,
}

impl DbsCapabilities {
	#[cfg(feature = "scripting")]
	fn get_scripting(&self) -> bool {
		// Even if there was a global deny, we allow if there is a specific allow for
		// scripting Even if there is a global allow, we deny if there is a specific
		// deny for scripting
		self.allow_scripting || (self.allow_all && !self.deny_scripting)
	}

	#[cfg(not(feature = "scripting"))]
	fn get_scripting(&self) -> bool {
		false
	}

	fn get_allow_guests(&self) -> bool {
		// Even if there was a global deny, we allow if there is a specific allow for
		// guests Even if there is a global allow, we deny if there is a specific deny
		// for guests
		self.allow_guests || (self.allow_all && !self.deny_guests)
	}

	fn get_allow_funcs(&self) -> Targets<FuncTarget> {
		// If there was a global deny, we allow if there is a general allow or some
		// specific allows for functions
		if self.deny_all {
			if let Some(targets) = &self.allow_funcs {
				match targets {
					Targets::None => {}
					Targets::Some(_) => {
						return targets.clone();
					}
					Targets::All => {
						return Targets::All;
					}
				}
			}
			return Targets::None;
		}

		// If there was a general deny for functions, we allow if there are specific
		// allows for functions
		if let Some(Targets::All) = self.deny_funcs {
			if let Some(targets) = &self.allow_funcs
				&& let Targets::Some(_) = targets
			{
				return targets.clone();
			}
			return Targets::None;
		}

		// If there are no high level denies but there is a global allow, we allow
		// functions
		if self.allow_all {
			return Targets::All;
		}

		// If there are no high level, we allow the provided functions
		// If nothing was provided, we allow functions by default (Targets::All)
		self.allow_funcs.clone().unwrap_or(Targets::All) // Functions are enabled by default for the server
	}

	fn get_allow_net(&self) -> Targets<NetTarget> {
		// If there was a global deny, we allow if there is a general allow or some
		// specific allows for networks
		if self.deny_all {
			if let Some(targets) = &self.allow_net {
				match targets {
					Targets::None => {}
					Targets::Some(_) => {
						return targets.clone();
					}
					Targets::All => {
						return Targets::All;
					}
				}
			}
			return Targets::None;
		}

		// If there was a general deny for networks, we allow if there are specific
		// allows for networks
		if let Some(Targets::All) = self.deny_net {
			if let Some(targets) = &self.allow_net
				&& let Targets::Some(_) = targets
			{
				return targets.clone();
			}
			return Targets::None;
		}

		// If there are no high level denies but there is a global allow, we allow
		// networks
		if self.allow_all {
			return Targets::All;
		}

		// If there are no high level denies, we allow the provided networks
		// If nothing was provided, we do not allow network by default (Targets::None)
		self.allow_net.clone().unwrap_or(Targets::None)
	}

	fn get_allow_rpc(&self) -> Targets<MethodTarget> {
		// If there was a global deny, we allow if there is a general allow or some
		// specific allows for RPC
		if self.deny_all {
			if let Some(targets) = &self.allow_rpc {
				match targets {
					Targets::None => {}
					Targets::Some(_) => {
						return targets.clone();
					}
					Targets::All => {
						return Targets::All;
					}
				}
			}
			return Targets::None;
		}

		// If there was a general deny for RPC, we allow if there are specific allows
		// for RPC methods
		if let Some(Targets::All) = self.deny_rpc {
			if let Some(targets) = self.allow_rpc.as_ref()
				&& let Targets::Some(_) = targets
			{
				return targets.clone();
			}
			return Targets::None;
		}

		// If there are no high level denies but there is a global allow, we allow RPC
		if self.allow_all {
			return Targets::All;
		}

		// If there are no high level denies, we allow the provided RPC methods
		// If nothing was provided, we allow RPC by default (Targets::All)
		self.allow_rpc.clone().unwrap_or(Targets::All) // RPC is enabled by default for the server
	}

	fn get_allow_http(&self) -> Targets<RouteTarget> {
		// If there was a global deny, we allow if there is a general allow or some
		// specific allows for HTTP
		if self.deny_all {
			if let Some(targets) = self.allow_http.as_ref() {
				match targets {
					Targets::None => {}
					Targets::Some(_) => {
						return targets.clone();
					}
					Targets::All => {
						return Targets::All;
					}
				}
			}
			return Targets::None;
		}

		// If there was a general deny for HTTP, we allow if there are specific allows
		// for HTTP routes
		if let Some(Targets::All) = self.deny_http {
			if let Some(targets) = self.allow_http.as_ref()
				&& let Targets::Some(_) = targets
			{
				return targets.clone();
			}
			return Targets::None;
		}

		// If there are no high level denies but there is a global allow, we allow HTTP
		if self.allow_all {
			return Targets::All;
		}

		// If there are no high level denies, we allow the provided HTTP routes
		// If nothing was provided, we allow HTTP by default (Targets::All)
		self.allow_http.clone().unwrap_or(Targets::All) // HTTP is enabled by default for the server
	}

	fn get_allow_experimental(&self) -> Targets<ExperimentalTarget> {
		// If there was a global deny, we allow if there is a general allow or some
		// specific allows for experimental features
		if self.deny_all {
			return self.allow_experimental.clone().unwrap_or(Targets::None);
		}

		// If there was a general deny for experimental features, we allow if there are
		// specific targets
		if let Some(Targets::All) = self.deny_experimental {
			match &self.allow_experimental {
				Some(t @ Targets::Some(_)) => return t.clone(),
				_ => return Targets::None,
			}
		}

		// If there are no high level denies, we allow the provided Experimental
		// features If nothing was provided, we deny Experimental targets by default
		// (Targets::None)
		self.allow_experimental.clone().unwrap_or(Targets::None) // Experimental targets are disabled by default for the server
	}

	fn get_allow_arbitrary_query(&self) -> Targets<ArbitraryQueryTarget> {
		// If there was a general deny for arbitrary queries, we allow if there are
		// specific allows for arbitrary query subjects
		if let Some(Targets::All) = self.deny_arbitrary_query {
			match &self.allow_arbitrary_query {
				Some(t @ Targets::Some(_)) => return t.clone(),
				_ => return Targets::None,
			}
		}

		// If there are no high level denies but there is a global allow, we allow
		// arbitrary queries
		if self.allow_all {
			return Targets::All;
		}

		// If there are no high level denies, we allow the provided arbitrary query
		// subjects If nothing was provided, we allow arbitrary queries by default
		// (Targets::All)
		self.allow_arbitrary_query.clone().unwrap_or(Targets::All) // arbitrary queries are enabled by default for the server
	}

	fn get_allow_eval_query(&self) -> Targets<EvalQueryTarget> {
		// If there was a general deny for eval queries, we allow only the
		// specific eval query subjects that are explicitly allowed.
		if let Some(Targets::All) = self.deny_eval_query {
			match &self.allow_eval_query {
				Some(t @ Targets::Some(_)) => return t.clone(),
				_ => return Targets::None,
			}
		}

		// Unlike arbitrary queries, eval is NOT enabled by `--allow-all`: it must
		// be explicitly enabled per subject group. If nothing was provided, eval
		// is denied for everyone (Targets::None).
		self.allow_eval_query.clone().unwrap_or(Targets::None)
	}

	fn get_deny_funcs(&self) -> Targets<FuncTarget> {
		// Allowed functions already consider a global deny and a general deny for
		// functions On top of what is explicitly allowed, we deny what is
		// specifically denied
		if let Some(targets) = &self.deny_funcs
			&& let Targets::Some(_) = targets
		{
			return targets.clone();
		}
		Targets::None
	}

	fn get_deny_net(&self) -> Targets<NetTarget> {
		// Allowed networks already consider a global deny and a general deny for
		// networks On top of what is explicitly allowed, we deny what is specifically
		// denied
		if let Some(targets) = &self.deny_net
			&& let Targets::Some(_) = targets
		{
			return targets.clone();
		}
		Targets::None
	}

	fn get_deny_all(&self) -> bool {
		self.deny_all
	}

	fn get_deny_rpc(&self) -> Targets<MethodTarget> {
		// Allowed RPC methods already consider a global deny and a general deny for RPC
		// On top of what is explicitly allowed, we deny what is specifically denied
		if let Some(targets) = &self.deny_rpc
			&& let Targets::Some(_) = targets
		{
			return targets.clone();
		}
		Targets::None
	}

	fn get_deny_http(&self) -> Targets<RouteTarget> {
		// Allowed HTTP routes already consider a global deny and a general deny for
		// HTTP On top of what is explicitly allowed, we deny what is specifically
		// denied
		if let Some(targets) = self.deny_http.as_ref()
			&& let Targets::Some(_) = targets
		{
			return targets.clone();
		}
		Targets::None
	}

	fn get_deny_experimental(&self) -> Targets<ExperimentalTarget> {
		// Allowed experimental targets already consider a global deny and a general
		// deny for experimental targets On top of what is explicitly allowed, we deny
		// what is specifically denied
		if let Some(t @ Targets::Some(_)) = &self.deny_experimental {
			t.clone()
		} else {
			Targets::None
		}
	}

	fn get_deny_arbitrary_query(&self) -> Targets<ArbitraryQueryTarget> {
		// Allowed arbitrary queryies already consider a global deny and a general deny
		// for arbitr On top of what is explicitly allowed, we deny what is
		// specifically denied
		if let Some(t @ Targets::Some(_)) = &self.deny_arbitrary_query {
			t.clone()
		} else {
			Targets::None
		}
	}

	fn get_deny_eval_query(&self) -> Targets<EvalQueryTarget> {
		// On top of what is explicitly allowed, we deny what is specifically denied.
		if let Some(t @ Targets::Some(_)) = &self.deny_eval_query {
			t.clone()
		} else {
			Targets::None
		}
	}

	pub fn into_cli_capabilities(self) -> Capabilities {
		merge_capabilities(SdkCapabilities::all().into(), &self)
	}
}

fn merge_capabilities(initial: Capabilities, caps: &DbsCapabilities) -> Capabilities {
	initial
		.with_scripting(caps.get_scripting())
		.with_guest_access(caps.get_allow_guests())
		.with_functions(caps.get_allow_funcs())
		.without_functions(caps.get_deny_funcs())
		.with_network_targets(caps.get_allow_net())
		.without_network_targets(caps.get_deny_net())
		.with_rpc_methods(caps.get_allow_rpc())
		.without_rpc_methods(caps.get_deny_rpc())
		.with_http_routes(caps.get_allow_http())
		.without_http_routes(caps.get_deny_http())
		.with_experimental(caps.get_allow_experimental())
		.without_experimental(caps.get_deny_experimental())
		.with_arbitrary_query(caps.get_allow_arbitrary_query())
		.without_arbitrary_query(caps.get_deny_arbitrary_query())
		.with_eval_query(caps.get_allow_eval_query())
		.without_eval_query(caps.get_deny_eval_query())
		.with_planner_strategy(caps.planner_strategy)
}

impl From<DbsCapabilities> for Capabilities {
	fn from(caps: DbsCapabilities) -> Self {
		merge_capabilities(Default::default(), &caps)
	}
}

/// Retry an async operation until it succeeds or the configured timeout is reached.
/// This is required for operations that rely on remote or distributed KV store
/// that may not be immediately available.
async fn retry_with_timeout<F, Fut, T, E>(
	operation_name: &str,
	timeout_duration: Duration,
	f: F,
) -> Result<T, anyhow::Error>
where
	F: Fn() -> Fut,
	Fut: Future<Output = Result<T, E>>,
	E: std::fmt::Display + std::fmt::Debug,
{
	retry_with_timeout_check(operation_name, timeout_duration, f, |_| false).await
}

/// Retry an async operation until it succeeds, the configured timeout is reached,
/// or a permanent error is detected.
///
/// # Parameters
/// - `operation_name`: Name of the operation for logging purposes
/// - `timeout_duration`: Total retry budget for this operation
/// - `f`: The async function to retry
/// - `is_permanent`: Predicate that returns `true` if an error is permanent and should not be
///   retried (e.g. storage version mismatch)
///
/// # Returns
/// The result of the operation if successful within the timeout
async fn retry_with_timeout_check<F, Fut, T, E, P>(
	operation_name: &str,
	timeout_duration: Duration,
	f: F,
	is_permanent: P,
) -> Result<T, anyhow::Error>
where
	F: Fn() -> Fut,
	Fut: Future<Output = Result<T, E>>,
	E: std::fmt::Display + std::fmt::Debug,
	P: Fn(&E) -> bool,
{
	let start = Instant::now();
	let mut attempt = 0;
	let mut last_reported = start;

	loop {
		attempt += 1;

		match timeout(timeout_duration.saturating_sub(start.elapsed()), f()).await {
			Ok(Ok(result)) => {
				if attempt > 1 {
					info!(target: TARGET, operation = operation_name, attempts = attempt, "Operation succeeded after retry");
				} else {
					info!(target: TARGET, operation = operation_name, attempts = attempt, "Operation succeeded");
				}
				return Ok(result);
			}
			Ok(Err(e)) => {
				// If this is a permanent error, fail immediately without retrying
				if is_permanent(&e) {
					error!(
						target: TARGET,
						operation = operation_name,
						error = %e,
						"Operation failed with a permanent error, not retrying"
					);
					return Err(anyhow::anyhow!("{e}"));
				}

				let elapsed = start.elapsed();
				if elapsed >= timeout_duration {
					return Err(anyhow::anyhow!(
						"Operation '{}' failed after {} attempts over {:?}: {}",
						operation_name,
						attempt,
						elapsed,
						e
					));
				}

				// The first failure always reports, and one more per interval
				// after that, so an operator watching a start that is waiting on
				// storage sees it without the log scaling with the retry rate.
				if attempt == 1 || last_reported.elapsed() >= STARTUP_RETRY_LOG_INTERVAL {
					last_reported = Instant::now();
					info!(
						target: TARGET,
						operation = operation_name,
						attempt = attempt,
						elapsed = ?elapsed,
						budget = ?timeout_duration,
						error = %e,
						"Operation failed, still retrying within the startup budget"
					);
				} else {
					debug!(
						target: TARGET,
						operation = operation_name,
						attempt = attempt,
						error = %e,
						"Operation failed, retrying..."
					);
				}

				// Exponential backoff with jitter: 100ms, 200ms, 400ms, 800ms, 1600ms, capped at 5s
				// Jitter adds randomness (0.5 to 1.5x) to prevent thundering herd
				let base_backoff = Duration::from_millis(100 * 2u64.pow((attempt - 1).min(5)));
				let base_backoff = base_backoff.min(Duration::from_secs(5));
				let jitter = rand::rng().random_range(0.5..=1.5);
				let backoff = base_backoff.mul_f64(jitter);
				sleep(backoff).await;
			}
			Err(_) => {
				return Err(anyhow::anyhow!(
					"Operation '{}' timed out after {} attempts over {:?}",
					operation_name,
					attempt,
					start.elapsed()
				));
			}
		}
	}
}

/// Whether a startup failure is a permanent property of the data on disk.
///
/// Each of these tells an operator to run a different build or restore an
/// export; none can be retried into success, and retrying only delays the
/// message reaching them. Matched on the rendered text because the errors cross
/// the core boundary as `anyhow::Error`.
///
/// `MigrationTimedOut` is deliberately absent: it means another live node is
/// still applying migrations, which is the one startup condition that a retry
/// can resolve on its own.
fn is_permanent_storage_error(e: &anyhow::Error) -> bool {
	let message = e.to_string();
	// `OutdatedStorageVersion`, `MigratedBeyondStorageVersion` and
	// `InvalidStorageVersion` respectively. The last reaches startup because the
	// version bootstrap declines to retry a marker it cannot decode; retrying it
	// here would spend the budget and then report a timeout, which names none of
	// what an operator has to act on.
	message.contains("out-of-date")
		|| message.contains("cannot be read by this one")
		|| message.contains("invalid storage version")
}

#[instrument(level = "trace", target = "surreal::dbs", skip_all)]
// The return tuple is the established shape of this internal startup
// entrypoint; the deferred startup work is just one more element on it.
#[allow(clippy::type_complexity)]
/// Initialise the database server
///
/// Creates and configures the datastore with the provided options.
///
/// # Parameters
/// - `factory`: Transaction builder factory for datastore backend selection
/// - `opt`: Server configuration including database path and authentication
/// - `canceller`: Token for graceful shutdown and cancellation of long-running operations
///
/// # Generic parameters
/// - `F`: Transaction builder factory type implementing `TransactionBuilderFactory`
pub async fn init<C: TransactionBuilderFactory>(
	composer: C,
	opt: &Config,
	canceller: CancellationToken,
	observer: Arc<dyn ExecutionObserver>,
	#[cfg_attr(not(storage), allow(unused_variables))] StartCommandDbsOptions {
		strict_mode,
		query_timeout,
		transaction_timeout,
		startup_operation_timeout,
		unauthenticated,
		capabilities,
		temporary_directory,
		import_file,
		slow_log_threshold,
		slow_log_param_allow,
		slow_log_param_deny,
		default_namespace,
		default_database,
		no_defaults,
		#[cfg_attr(not(feature = "surrealism"), allow(unused_variables))]
		lazy_surrealism,
	}: StartCommandDbsOptions,
) -> Result<(Arc<Datastore>, Receiver<Notification>, C::RouterState, PendingStartup)> {
	// Warn about the strict mode flag being unused.
	if let Some(true) = strict_mode {
		warn!(
			"Strict mode is no longer defined on the server level. Use `DEFINE DATABASE <db> STRICT` instead. Ignoring strict mode flag."
		);
	}
	// Log specified query timeout
	if let Some(v) = query_timeout {
		debug!("Maximum query processing timeout is {v:?}");
	}
	// Log specified parse timeout
	if let Some(v) = transaction_timeout {
		debug!("Maximum transaction processing timeout is {v:?}");
	}
	debug!("Startup operation timeout is {startup_operation_timeout:?}");
	// Log whether authentication is disabled
	if unauthenticated {
		warn!(
			"❌🔒 IMPORTANT: Authentication is disabled. This is not recommended for production use. 🔒❌"
		);
	}
	// Warn about the impact of denying all capabilities
	if capabilities.get_deny_all() {
		warn!(
			"You are denying all capabilities by default. Although this is recommended, beware that any new capabilities will also be denied."
		);
	}
	if let Some(v) = slow_log_threshold {
		debug!("Slow log threshold is {v:?}");
	}
	if !slow_log_param_allow.is_empty() {
		debug!("Slow log param allow is {:?}", slow_log_param_allow);
	}
	if !slow_log_param_deny.is_empty() {
		debug!("Slow log param deny is {:?}", slow_log_param_deny);
	}
	// Convert the capabilities
	let capabilities = capabilities.into();
	// Log the specified server capabilities
	debug!("Server capabilities: {capabilities}");

	let (send, recv) = surrealdb_core::channel::bounded(surrealdb_cnf::NOTIFICATIONS_CHANNEL_SIZE);

	let config = ConfigMap::from_env();
	// Parse and setup the desired kv datastore
	let builder = Datastore::builder()
		// The cadences the maintenance tasks run at, once this function starts
		// them below.
		.with_engine_options(opt.engine)
		// Started by hand from `initialise_datastore`, after `check_version`,
		// rather than at construction. Every one of these tasks writes, and a
		// write that lands before the version key does makes a datastore this
		// process just created look like one written before that key existed —
		// which `check_version` is obliged to refuse, permanently, on this and
		// every later start.
		.without_maintenance_tasks()
		.with_config(config)
		// Mirror the tokio runtime worker count the server built itself
		// with (see `cnf::RUNTIME_WORKER_THREADS`) into the datastore
		// config so the RocksDB engine sizes its inline-blocking permit
		// cap from the actual executor width.
		.with_runtime_worker_threads(*crate::cnf::RUNTIME_WORKER_THREADS)
		.with_query_timeout(query_timeout)
		.with_transaction_timeout(transaction_timeout)
		.with_auth(!unauthenticated)
		.with_capabilities(capabilities)
		.with_notify(send)
		.with_shutdown_cancel(canceller.clone())
		.with_observer(observer);

	#[cfg(storage)]
	let builder = builder.with_temporary_directory(temporary_directory);

	let builder = if let Some(slow_log_threshold) = slow_log_threshold {
		builder.with_slow_log(slow_log_threshold, slow_log_param_allow, slow_log_param_deny)
	} else {
		builder
	};

	#[cfg(feature = "surrealism")]
	let builder = builder.with_lazy_surrealism(lazy_surrealism);

	let (dbs, router_state) =
		builder.build_with_factory_path_and_router_state::<C>(&opt.path, composer).await?;
	// Building the datastore is all this function does: every startup step that
	// opens a transaction is handed back for the caller to run through
	// [`initialise_datastore`] and [`finish_startup`] once the HTTP listener is
	// bound.
	//
	// This is what lets a composer hand back a datastore that cannot serve a
	// transaction yet. Nothing here opens one, so a composer whose engine has to
	// join a cluster no longer has to finish joining before it returns: it can
	// hand back the builder and let the join finish behind the readiness gate,
	// which is the only way `/status` and `/ready` can answer during a join.
	// Until such a composer does so, the wait simply moves — this function still
	// cannot return before the builder it is given exists.
	//
	// It also keeps the steps off the bind on every other engine, where they are
	// fast but not free, and where storage that is slow to answer would
	// otherwise hold the listener for the whole retry budget.
	let pending_startup = PendingStartup {
		import_file,
		credentials: opt.user.clone().zip(opt.pass.clone()),
		default_namespace,
		default_database,
		no_defaults,
		engine: opt.engine,
		canceller,
		timeout: startup_operation_timeout,
	};
	// All ok
	Ok((dbs, recv, router_state, pending_startup))
}

/// Starts the first peer-expiry and archived-node cleanup pass in the
/// background.
///
/// Ordering is the whole point of running this from here rather than leaving it
/// to the maintenance scheduler's own cadence:
///
/// - It runs after `check_version`, so a pass can never commit catalog mutations to a datastore
///   whose version validation goes on to refuse startup. The scheduler is held back for the same
///   reason (see the `without_maintenance_tasks` builder call in `init`), but this pass has to be
///   ordered explicitly because it does not wait for the first tick.
/// - It runs after this node's own `insert_node`, so the id this process is about to serve under is
///   already active when the pass takes its snapshot. A node id that is stable across restarts
///   would otherwise be visible to an earlier pass as its previous incarnation's stale or archived
///   row, and both jobs decide from a snapshot taken before their write transaction.
///
/// It is spawned rather than awaited because it must not gate readiness:
/// `remove_nodes` opens one write transaction per archived node, so its cost
/// scales with however much dead-member residue a restarting cluster left
/// behind, with no bound.
///
/// Expiry runs before cleanup, so a peer that has just timed out is archived in
/// time for the same pass to reap it. Either step is skipped when its interval
/// is zero, which is how both are disabled — the same contract the scheduler's
/// own registration applies. Failures are logged and do not stop the other
/// step: the periodic `NodeExpire` / `NodeCleanup` jobs run on their normal
/// cadence from here on and pick up whatever this pass left.
fn spawn_first_node_maintenance_pass(
	dbs: &Arc<Datastore>,
	engine: &EngineOptions,
	canceller: CancellationToken,
) {
	let expire = !engine.node_membership_check_interval.is_zero();
	let cleanup = !engine.node_membership_cleanup_interval.is_zero();
	if !expire && !cleanup {
		return;
	}
	/// Reports a failed step, unless the server is shutting down: the storage
	/// engine refuses commits from that point on, so a failure there describes
	/// the shutdown rather than a fault.
	fn report(canceller: &CancellationToken, step: &str, res: Result<()>) {
		if let Err(err) = res
			&& !canceller.is_cancelled()
		{
			warn!(target: TARGET, "Error {step}: {err}");
		}
	}

	let dbs = Arc::clone(dbs);
	tokio::spawn(async move {
		if expire && !canceller.is_cancelled() {
			report(&canceller, "expiring inactive nodes", dbs.expire_nodes().await);
		}
		if cleanup && !canceller.is_cancelled() {
			report(&canceller, "removing archived nodes", dbs.remove_nodes().await);
		}
	});
}

/// Startup work deferred out of [`init`] so the caller can bind the HTTP
/// listener first and run it concurrently (via [`initialise_datastore`], then
/// [`finish_startup`]). Until both have completed the instance is not ready to
/// serve queries (see the readiness gate in `ntw`).
pub(crate) struct PendingStartup {
	/// SurrealQL file to import on startup, if one was provided.
	import_file: Option<PathBuf>,
	/// Root credentials (user, pass) to create if no root user already exists.
	credentials: Option<(String, String)>,
	/// Namespace to create on a datastore that turns out to be new; `main` when
	/// unset.
	default_namespace: Option<String>,
	/// Database to create on a datastore that turns out to be new; `main` when
	/// unset.
	default_database: Option<String>,
	/// Whether to skip creating the default namespace and database entirely.
	no_defaults: bool,
	/// Cadences the maintenance schedule and its first node pass run at.
	engine: EngineOptions,
	/// Cancelled when the server shuts down, so the first node-maintenance pass
	/// stops rather than writing into a closing storage engine.
	canceller: CancellationToken,
	/// Per-operation retry budget (from `SURREAL_STARTUP_OPERATION_TIMEOUT`).
	timeout: Duration,
}

impl PendingStartup {
	/// Whether a startup import is part of the deferred work.
	///
	/// The caller uses this to decide whether the HTTP bind waits for that work:
	/// an import is unbounded, so waiting for one would starve the probes for as
	/// long as it runs, which is what deferring it avoids.
	pub(crate) fn has_import(&self) -> bool {
		self.import_file.is_some()
	}
}

/// Bring the datastore built by [`init`] into the state a server can serve
/// from: stamp the storage version, create the configured defaults if the
/// datastore is new, register this node's cluster membership row, and start the
/// maintenance schedule.
///
/// None of it can be skipped, so a failure here is fatal to the instance rather
/// than something it can serve without — the caller is expected to end the
/// process rather than leave a server that can never become ready.
///
/// The order is load-bearing:
///
/// - `check_version` first, and the maintenance schedule only after it. Every maintenance task
///   writes, and a write that lands before the version key does makes a datastore this process just
///   created look like one written before that key existed — which `check_version` is obliged to
///   refuse, permanently, on this and every later start.
/// - The storage-state conditions `check_version` reports are permanent — the data on disk is from
///   another version — so they are not retried: retrying them would waste the budget and delay the
///   pod restart that operators need the failure to trigger.
/// - `insert_node` before the first node-maintenance pass, so the id this process is about to serve
///   under is already active when that pass takes its snapshot (see
///   [`spawn_first_node_maintenance_pass`]).
/// - The wait for the backend to become serve-ready comes before all of it, and is not part of the
///   retry budget the steps share.
pub(crate) async fn initialise_datastore(
	dbs: &Arc<Datastore>,
	pending: &PendingStartup,
) -> Result<()> {
	// A backend that has to join a cluster before it can answer reports that
	// wait here. It is unbounded on purpose: every step below opens a
	// transaction under the shared `pending.timeout` budget, and a join that
	// legitimately takes longer than that budget would otherwise exhaust it and
	// fail a start that was only ever going to be slow. The readiness gate
	// covers the window, and the caller cancels this task at shutdown.
	dbs.wait_until_serve_ready().await?;
	// Ensure the storage version is up to date to prevent corruption.
	let (_, is_new) = retry_with_timeout_check(
		"check_version",
		pending.timeout,
		|| async { dbs.check_version().await },
		is_permanent_storage_error,
	)
	.await?;
	// Create default namespace and database if not disabled
	if is_new && !pending.no_defaults {
		let ns = pending.default_namespace.as_deref().unwrap_or("main");
		let db = pending.default_database.as_deref().unwrap_or("main");
		// Initialise defaults
		retry_with_timeout("initialise_defaults", pending.timeout, || async {
			dbs.initialise_defaults(ns, db).await
		})
		.await?;
	}
	// Register this node's cluster membership row. Until it exists this node has
	// no heartbeat, which `/ready` reports as still starting.
	retry_with_timeout("Insert node", pending.timeout, || async { dbs.insert_node().await })
		.await?;
	// The version is stamped and this node is registered, so the maintenance
	// tasks may start writing.
	dbs.start_maintenance_tasks();
	spawn_first_node_maintenance_pass(dbs, &pending.engine, pending.canceller.clone());
	Ok(())
}

/// Run the deferred startup content: execute the import file (if any), then
/// initialise the root credentials (if provided). Runs after
/// [`initialise_datastore`], which is what makes the datastore writable.
///
/// The import runs *before* credential initialisation so that an import which
/// supplies root users suppresses creation of the CLI bootstrap user, matching
/// the historical (pre-early-bind) ordering.
pub(crate) async fn finish_startup(ds: &Datastore, pending: &PendingStartup) -> Result<()> {
	// Import the startup file.
	//
	// An import is not transactional: each statement commits on its own and
	// execution carries on past a failure, so the result of the call says only
	// that the dump parsed. Every statement that did not apply is logged here,
	// because startup proceeds either way and this is the operator's only
	// account of what the file actually restored.
	//
	// The failures are not returned: `retry_with_timeout` retries on `Err`, and
	// re-running a dump that has already applied part of itself is worse than
	// booting with the loss reported.
	if let Some(file) = &pending.import_file {
		info!(target: TARGET, file = ?file, "Importing data from file");
		let sql = fs::read_to_string(file)?;
		let results = retry_with_timeout("startup", pending.timeout, || async {
			ds.startup(&sql, &Session::owner()).await
		})
		.await?;
		let failed = results.iter().filter(|r| r.result.is_err()).count();
		if failed > 0 {
			error!(
				target: TARGET,
				file = ?file,
				failed,
				total = results.len(),
				"Startup import did not apply in full; the database is partially restored"
			);
			for result in results.iter() {
				if let Err(err) = &result.result {
					error!(target: TARGET, error = %err, "Startup import statement failed");
				}
			}
		}
	}
	// Create the CLI-provided root user if no root user exists yet.
	if let Some((user, pass)) = &pending.credentials {
		info!(target: TARGET, user = %user, "Initialising credentials");
		retry_with_timeout("initialise_credentials", pending.timeout, || async {
			ds.initialise_credentials(user, pass).await
		})
		.await?;
	}
	Ok(())
}

#[cfg(test)]
mod tests {
	use std::ffi::OsString;
	use std::str::FromStr;
	use std::sync::Arc;
	use std::sync::atomic::{AtomicUsize, Ordering};

	use clap::Parser;
	use serial_test::serial;
	// Named here rather than inside each test because the shared helpers below
	// carry them in their signatures.
	#[cfg(feature = "storage-rocksdb")]
	use surrealdb_datastore::key::schema::{NodeKey, NodeLiveQueryKey};
	use surrealdb_types::ToSql;
	use test_log::test;
	use wiremock::matchers::{method, path};
	use wiremock::{Mock, MockServer, ResponseTemplate};

	use super::*;

	const STARTUP_OPERATION_TIMEOUT_ENV: &str = "SURREAL_STARTUP_OPERATION_TIMEOUT";

	struct EnvGuard {
		key: &'static str,
		old: Option<OsString>,
	}

	impl EnvGuard {
		fn set(key: &'static str, value: &str) -> Self {
			let old = std::env::var_os(key);
			// SAFETY: These tests mutate a single process environment variable and are
			// serialized so no other test in this module observes the temporary value.
			unsafe {
				std::env::set_var(key, value);
			}
			Self {
				key,
				old,
			}
		}

		fn remove(key: &'static str) -> Self {
			let old = std::env::var_os(key);
			// SAFETY: These tests mutate a single process environment variable and are
			// serialized so no other test in this module observes the temporary value.
			unsafe {
				std::env::remove_var(key);
			}
			Self {
				key,
				old,
			}
		}
	}

	impl Drop for EnvGuard {
		fn drop(&mut self) {
			// SAFETY: The guard restores the serialized test's temporary environment change.
			unsafe {
				match &self.old {
					Some(value) => std::env::set_var(self.key, value),
					None => std::env::remove_var(self.key),
				}
			}
		}
	}

	#[derive(Parser, Debug)]
	struct TestCli {
		#[command(flatten)]
		dbs: StartCommandDbsOptions,
	}

	#[test]
	#[serial]
	fn startup_operation_timeout_defaults_to_sixty_seconds() {
		let _guard = EnvGuard::remove(STARTUP_OPERATION_TIMEOUT_ENV);
		let cli = TestCli::try_parse_from(["surrealdb"]).unwrap();
		assert_eq!(cli.dbs.startup_operation_timeout, Duration::from_secs(60));
	}

	#[test]
	fn startup_operation_timeout_can_be_set_from_cli() {
		let cli =
			TestCli::try_parse_from(["surrealdb", "--startup-operation-timeout", "10m"]).unwrap();
		assert_eq!(cli.dbs.startup_operation_timeout, Duration::from_secs(10 * 60));
	}

	#[test]
	#[serial]
	fn startup_operation_timeout_can_be_set_from_env() {
		let _guard = EnvGuard::set(STARTUP_OPERATION_TIMEOUT_ENV, "75s");
		let cli = TestCli::try_parse_from(["surrealdb"]).unwrap();
		assert_eq!(cli.dbs.startup_operation_timeout, Duration::from_secs(75));
	}

	#[test(tokio::test(start_paused = true))]
	async fn startup_retry_times_out_attempt_without_retrying() {
		let attempts = Arc::new(AtomicUsize::new(0));

		let err = retry_with_timeout("test operation", Duration::from_millis(10), || {
			let attempts = Arc::clone(&attempts);
			async move {
				attempts.fetch_add(1, Ordering::SeqCst);
				sleep(Duration::from_millis(50)).await;
				Ok::<_, &'static str>(())
			}
		})
		.await
		.unwrap_err();

		assert!(err.to_string().contains("timed out after 1 attempts"));
		assert_eq!(attempts.load(Ordering::SeqCst), 1);
	}

	#[test(tokio::test(start_paused = true))]
	async fn startup_retry_returns_permanent_errors_immediately() {
		let attempts = Arc::new(AtomicUsize::new(0));

		let err = retry_with_timeout_check(
			"test operation",
			Duration::from_millis(100),
			|| {
				let attempts = Arc::clone(&attempts);
				async move {
					attempts.fetch_add(1, Ordering::SeqCst);
					Err::<(), _>("permanent")
				}
			},
			|e| *e == "permanent",
		)
		.await
		.unwrap_err();

		assert_eq!(err.to_string(), "permanent");
		assert_eq!(attempts.load(Ordering::SeqCst), 1);
	}

	/// Every storage-state error the version bootstrap can raise is recognised
	/// as permanent by the server's own classifier.
	///
	/// The classifier matches rendered text, because these cross the core
	/// boundary as `anyhow::Error` and the type is not nameable here. That makes
	/// it drift silently when a variant is reworded or a new one starts reaching
	/// startup, so the fixtures are the real errors rather than copies of their
	/// wording: a reword that breaks the match breaks this too.
	#[test]
	fn every_storage_state_error_is_classified_permanent() {
		use surrealdb_datastore::DatastoreError;

		for error in [
			DatastoreError::InvalidStorageVersion,
			DatastoreError::OutdatedStorageVersion {
				expected: 3,
				actual: 1,
			},
			DatastoreError::MigratedBeyondStorageVersion {
				stored: "3.4.0".to_string(),
				running: "3.3.0".to_string(),
				migrations: "2".to_string(),
			},
		] {
			let rendered = error.to_string();
			assert!(
				is_permanent_storage_error(&anyhow::Error::new(error)),
				"a permanent storage-state error would be retried: {rendered}"
			);
		}
	}

	/// A migration another node is still applying is the one startup condition a
	/// retry resolves on its own, so it must stay retryable.
	#[test]
	fn a_pending_migration_is_not_classified_permanent() {
		use surrealdb_datastore::DatastoreError;

		assert!(!is_permanent_storage_error(&anyhow::Error::new(
			DatastoreError::MigrationTimedOut {
				migrations: "2".to_string(),
			}
		)));
	}

	#[test(tokio::test(flavor = "multi_thread"))]
	async fn test_capabilities() {
		let server1 = {
			let s = MockServer::start().await;
			let get = Mock::given(method("GET"))
				.and(path("/"))
				.respond_with(ResponseTemplate::new(200).set_body_string("SUCCESS"))
				.expect(1);

			let get2 = Mock::given(method("GET"))
				.and(path("/test"))
				.respond_with(ResponseTemplate::new(200).set_body_string("SUCCESS"))
				.expect(1);

			s.register(get).await;
			s.register(get2).await;
			s
		};

		let server2 = {
			let s = MockServer::start().await;
			let get = Mock::given(method("GET"))
				.respond_with(ResponseTemplate::new(200).set_body_string("SUCCESS"))
				.expect(1);
			let head =
				Mock::given(method("HEAD")).respond_with(ResponseTemplate::new(200)).expect(0);

			s.register(get).await;
			s.register(head).await;

			s
		};

		let server3 = {
			let s = MockServer::start().await;
			let redirect_res = ResponseTemplate::new(301).append_header("Location", server1.uri());

			let redirect = Mock::given(method("GET"))
				.and(path("redirect"))
				.respond_with(redirect_res)
				.expect(1);

			s.register(redirect).await;
			s
		};

		// (Datastore, Session, Query, Succeeds, Response Contains)
		let cases = vec![
			//
			// 0 - Functions and Networking are allowed
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::All),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				format!("RETURN http::get('{}')", server1.uri()),
				true,
				"SUCCESS".to_string(),
			),
			//
			// 1 - Scripting is allowed
			(
				Datastore::builder()
					.with_capabilities(Capabilities::default().with_scripting(true))
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				"RETURN function() { return '1' }".to_string(),
				true,
				"1".to_string(),
			),
			//
			// 2 - Scripting is not allowed
			(
				Datastore::builder()
					.with_capabilities(Capabilities::default().with_scripting(false))
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				"RETURN function() { return '1' }".to_string(),
				false,
				"Scripting functions are not allowed".to_string(),
			),
			//
			// 3 - Anonymous actor when guest access is allowed and auth is enabled, succeeds
			(
				Datastore::builder()
					.with_capabilities(Capabilities::default().with_guest_access(true))
					.with_auth(true)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::default(),
				"RETURN 1".to_string(),
				true,
				"1".to_string(),
			),
			//
			// 4 - Anonymous actor when guest access is not allowed and auth is enabled, throws
			// error
			(
				Datastore::builder()
					.with_capabilities(Capabilities::default().with_guest_access(false))
					.with_auth(true)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::default(),
				"RETURN 1".to_string(),
				false,
				"Not enough permissions to perform this action".to_string(),
			),
			//
			// 5 - Anonymous actor when guest access is not allowed and auth is disabled, succeeds
			(
				Datastore::builder()
					.with_auth(false)
					.with_capabilities(Capabilities::default().with_guest_access(false))
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::default(),
				"RETURN 1".to_string(),
				true,
				"1".to_string(),
			),
			//
			// 6 - Authenticated user when guest access is not allowed and auth is enabled,
			// succeeds
			(
				Datastore::builder()
					.with_auth(true)
					.with_capabilities(Capabilities::default().with_guest_access(false))
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::viewer(),
				"RETURN 1".to_string(),
				true,
				"1".to_string(),
			),
			// 7 - Specific experimental feature enabled
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default().with_experimental(ExperimentalTarget::Files.into()),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner().with_ns("test").with_db("test"),
				"DEFINE BUCKET test BACKEND \"memory\";".to_string(),
				true,
				"NONE".to_string(),
			),
			// 8 - Specific experimental feature disabled
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.without_experimental(ExperimentalTarget::Files.into()),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner().with_ns("test").with_db("test"),
				"DEFINE BUCKET test BACKEND \"memory\";".to_string(),
				false,
				"expected the experimental files feature to be enabled".to_string(),
			),
			//
			// 9 - Some functions are not allowed
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::Some(
								[FuncTarget::from_str("string::*").unwrap()].into(),
							))
							.without_functions(Targets::<FuncTarget>::Some(
								[FuncTarget::from_str("string::len").unwrap()].into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				"RETURN string::len('a')".to_string(),
				false,
				"Function 'string::len' is not allowed".to_string(),
			),
			// 10 -
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::Some(
								[FuncTarget::from_str("string::*").unwrap()].into(),
							))
							.without_functions(Targets::<FuncTarget>::Some(
								[FuncTarget::from_str("string::len").unwrap()].into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				"RETURN string::lowercase('A')".to_string(),
				true,
				"a".to_string(),
			),
			// 11 -
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::Some(
								[FuncTarget::from_str("string::*").unwrap()].into(),
							))
							.without_functions(Targets::<FuncTarget>::Some(
								[FuncTarget::from_str("string::len").unwrap()].into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				"RETURN time::now()".to_string(),
				false,
				"Function 'time::now' is not allowed".to_string(),
			),
			//
			// 12 - Some net targets are not allowed
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::Some(
								[
									NetTarget::from_str(&server1.address().to_string()).unwrap(),
									NetTarget::from_str(&server2.address().to_string()).unwrap(),
								]
								.into(),
							))
							.without_network_targets(Targets::<NetTarget>::Some(
								[NetTarget::from_str(&server1.address().to_string()).unwrap()]
									.into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				format!("RETURN http::get('{}')", server1.uri()),
				false,
				format!("Access to network target '{}' is not allowed", server1.address()),
			),
			// 13 -
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::Some(
								[
									NetTarget::from_str(&server1.address().to_string()).unwrap(),
									NetTarget::from_str(&server2.address().to_string()).unwrap(),
								]
								.into(),
							))
							.without_network_targets(Targets::<NetTarget>::Some(
								[NetTarget::from_str(&server1.address().to_string()).unwrap()]
									.into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				"RETURN http::get('http://1.1.1.1')".to_string(),
				false,
				"Access to network target '1.1.1.1:80' is not allowed".to_string(),
			),
			// 14 -
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::Some(
								[
									NetTarget::from_str(&server1.address().to_string()).unwrap(),
									NetTarget::from_str(&server2.address().to_string()).unwrap(),
								]
								.into(),
							))
							.without_network_targets(Targets::<NetTarget>::Some(
								[NetTarget::from_str(&server1.address().to_string()).unwrap()]
									.into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				format!("RETURN http::get('{}')", server2.uri()),
				true,
				"SUCCESS".to_string(),
			),
			(
				// 15 - Ensure redirect fails
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::Some(
								[NetTarget::from_str(&server3.address().to_string()).unwrap()]
									.into(),
							))
							.without_network_targets(Targets::<NetTarget>::Some(
								[NetTarget::from_str(&server1.address().to_string()).unwrap()]
									.into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				format!("RETURN http::get('{}/redirect')", server3.uri()),
				false,
				format!(
					"There was an error processing a remote HTTP request: error following redirect for url ({}/redirect)",
					server3.uri()
				),
			),
			(
				// 16 - Ensure connecting via localhost succeed
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::All),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				format!("RETURN http::get('http://localhost:{}/test')", server1.address().port()),
				true,
				"SUCCESS".to_string(),
			),
			// - 17
			(
				// Ensure connecting via localhost is denied when all IPs are blocked
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::All)
							.without_network_targets(Targets::<NetTarget>::Some(
								[
									NetTarget::from_str("127.0.0.1/0").unwrap(),
									NetTarget::from_str("::/0").unwrap(),
								]
								.into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				format!("RETURN http::get('http://localhost:{}')", server1.address().port()),
				false,
				"is not allowed".to_string(),
			),
			// 18 - Ensure redirect succeed
			(
				Datastore::builder()
					.with_capabilities(
						Capabilities::default()
							.with_functions(Targets::<FuncTarget>::All)
							.with_network_targets(Targets::<NetTarget>::Some(
								[NetTarget::from_str("github.com").unwrap()].into(),
							))
							.without_network_targets(Targets::<NetTarget>::Some(
								[
									NetTarget::from_str("0.0.0.0/8").unwrap(),
									NetTarget::from_str("10.0.0.0/8").unwrap(),
									NetTarget::from_str("10.18.0.0/16").unwrap(),
									NetTarget::from_str("10.2.0.0/16").unwrap(),
									NetTarget::from_str("100.64.0.0/10").unwrap(),
									NetTarget::from_str("127.0.0.0/8").unwrap(),
									NetTarget::from_str("169.254.0.0/16").unwrap(),
									NetTarget::from_str("172.16.0.0/12").unwrap(),
									NetTarget::from_str("172.20.0.0/16").unwrap(),
									NetTarget::from_str("192.0.0.0/24").unwrap(),
									NetTarget::from_str("192.168.0.0/16").unwrap(),
									NetTarget::from_str("192.88.99.0/24").unwrap(),
									NetTarget::from_str("198.18.0.0/15").unwrap(),
									NetTarget::from_str("::1/128").unwrap(),
									NetTarget::from_str("fc00::/7").unwrap(),
									NetTarget::from_str("fc00::/8").unwrap(),
								]
								.into(),
							)),
					)
					.build_with_path("memory")
					.await
					.unwrap(),
				Session::owner(),
				// This will be redirected to: https://github.com/surrealdb/surrealdb/pull/6293
				"RETURN http::get('https://github.com/surrealdb/surrealdb/issues/6293')"
					.to_string(),
				true,
				"<!DOCTYPE html>".to_string(),
			),
		];

		for (idx, (ds, sess, query, succeeds, contains)) in cases.into_iter().enumerate() {
			info!("Test case {idx}: query={query}, succeeds={succeeds}");
			let res = ds.execute(&query, &sess, None).await;

			if !succeeds && res.is_err() {
				let res = res.unwrap_err();
				assert!(
					res.to_string().contains(&contains),
					"Unexpected error for test case {}: {:?}",
					idx,
					res.to_string()
				);
				continue;
			}

			let res = res.unwrap().remove(0).output();
			let res = if succeeds {
				assert!(res.is_ok(), "Unexpected error for test case {idx}: {res:?}");
				res.unwrap().to_sql()
			} else {
				assert!(res.is_err(), "Unexpected success for test case {idx}: {res:?}");
				res.unwrap_err().to_string()
			};

			assert!(
				res.contains(&contains),
				"Unexpected result for test case {idx}: expected to contain = `{contains}`, got `{res}`"
			);
		}

		server1.verify().await;
		server2.verify().await;
		server3.verify().await;
	}

	#[test]
	fn test_dbs_capabilities_target_all() {
		let caps = DbsCapabilities {
			allow_all: false,
			#[cfg(feature = "scripting")]
			allow_scripting: false,
			allow_guests: false,
			allow_funcs: None,
			allow_experimental: Some(Targets::All),
			allow_arbitrary_query: Some(Targets::All),
			allow_eval_query: None,
			allow_net: None,
			allow_rpc: None,
			allow_http: None,
			deny_all: false,
			#[cfg(feature = "scripting")]
			deny_scripting: false,
			deny_guests: false,
			deny_funcs: None,
			deny_experimental: None,
			deny_arbitrary_query: None,
			deny_eval_query: None,
			deny_net: None,
			deny_rpc: None,
			deny_http: None,
			planner_strategy: NewPlannerStrategy::default(),
		};
		assert_eq!(caps.get_allow_experimental(), Targets::All);
		assert_eq!(caps.get_allow_arbitrary_query(), Targets::All);
	}

	#[test]
	fn test_dbs_capabilities_eval_query_denied_by_default() {
		// Even with allow_all set, eval is denied unless explicitly enabled, and
		// arbitrary-query (which it is also gated by) stays enabled.
		let mut caps = DbsCapabilities {
			allow_all: true,
			#[cfg(feature = "scripting")]
			allow_scripting: false,
			allow_guests: false,
			allow_funcs: None,
			allow_experimental: None,
			allow_arbitrary_query: None,
			allow_eval_query: None,
			allow_net: None,
			allow_rpc: None,
			allow_http: None,
			deny_all: false,
			#[cfg(feature = "scripting")]
			deny_scripting: false,
			deny_guests: false,
			deny_funcs: None,
			deny_experimental: None,
			deny_arbitrary_query: None,
			deny_eval_query: None,
			deny_net: None,
			deny_rpc: None,
			deny_http: None,
			planner_strategy: NewPlannerStrategy::default(),
		};
		assert_eq!(caps.get_allow_arbitrary_query(), Targets::All);
		assert_eq!(caps.get_allow_eval_query(), Targets::None, "eval is not enabled by allow_all");

		// Explicitly enabling a subject group turns it on.
		caps.allow_eval_query = Some(Targets::from(EvalQueryTarget::System));
		assert_eq!(caps.get_allow_eval_query(), Targets::from(EvalQueryTarget::System));
	}

	#[test]
	fn test_dbs_capabilities_deny_all_vs_allow_all() {
		// Test case 1: When deny_funcs = Targets::All and allow_funcs = Targets::All,
		// the deny should take precedence and return Targets::None
		let caps = DbsCapabilities {
			allow_all: false,
			#[cfg(feature = "scripting")]
			allow_scripting: false,
			allow_guests: false,
			allow_funcs: Some(Targets::All),
			allow_experimental: None,
			allow_arbitrary_query: None,
			allow_eval_query: None,
			allow_net: None,
			allow_rpc: None,
			allow_http: None,
			deny_all: false,
			#[cfg(feature = "scripting")]
			deny_scripting: false,
			deny_guests: false,
			deny_funcs: Some(Targets::All),
			deny_experimental: None,
			deny_arbitrary_query: None,
			deny_eval_query: None,
			deny_net: None,
			deny_rpc: None,
			deny_http: None,
			planner_strategy: NewPlannerStrategy::default(),
		};
		assert_eq!(
			caps.get_allow_funcs(),
			Targets::None,
			"When deny_funcs=All and allow_funcs=All, should deny (return None)"
		);

		// Test case 2: When deny_http = Targets::All and allow_http = Targets::All,
		// the deny should take precedence and return Targets::None
		let caps = DbsCapabilities {
			allow_all: false,
			#[cfg(feature = "scripting")]
			allow_scripting: false,
			allow_guests: false,
			allow_funcs: None,
			allow_experimental: None,
			allow_arbitrary_query: None,
			allow_eval_query: None,
			allow_net: None,
			allow_rpc: None,
			allow_http: Some(Targets::All),
			deny_all: false,
			#[cfg(feature = "scripting")]
			deny_scripting: false,
			deny_guests: false,
			deny_funcs: None,
			deny_experimental: None,
			deny_arbitrary_query: None,
			deny_eval_query: None,
			deny_net: None,
			deny_rpc: None,
			deny_http: Some(Targets::All),
			planner_strategy: NewPlannerStrategy::default(),
		};
		assert_eq!(
			caps.get_allow_http(),
			Targets::None,
			"When deny_http=All and allow_http=All, should deny (return None)"
		);

		// Test case 3: When deny_funcs = Targets::All and allow_funcs = Targets::None,
		// should return Targets::None
		let caps = DbsCapabilities {
			allow_all: false,
			#[cfg(feature = "scripting")]
			allow_scripting: false,
			allow_guests: false,
			allow_funcs: Some(Targets::None),
			allow_experimental: None,
			allow_arbitrary_query: None,
			allow_eval_query: None,
			allow_net: None,
			allow_rpc: None,
			allow_http: None,
			deny_all: false,
			#[cfg(feature = "scripting")]
			deny_scripting: false,
			deny_guests: false,
			deny_funcs: Some(Targets::All),
			deny_experimental: None,
			deny_arbitrary_query: None,
			deny_eval_query: None,
			deny_net: None,
			deny_rpc: None,
			deny_http: None,
			planner_strategy: NewPlannerStrategy::default(),
		};
		assert_eq!(
			caps.get_allow_funcs(),
			Targets::None,
			"When deny_funcs=All and allow_funcs=None, should deny (return None)"
		);
	}

	/// Seeds a fresh datastore at `path` with an archived peer node and one of
	/// the live-query records it owns, then closes it so the caller can open the
	/// same path through [`init`]. Returns the keys of the two seeded rows.
	///
	/// The heartbeat sits at the epoch, so no reading of the clock makes the row
	/// look live, and the seeding datastore runs no maintenance tasks, so nothing
	/// reaps the rows before the caller reopens the path.
	#[cfg(feature = "storage-rocksdb")]
	async fn seed_archived_peer(path: &str) -> Result<(NodeKey, NodeLiveQueryKey)> {
		use surrealdb_core::dbs::node::{Node, Timestamp};
		use surrealdb_datastore::TransactionType;
		use surrealdb_datastore::catalog::{DatabaseId, NamespaceId, NodeLiveQuery};
		use uuid::Uuid;

		let peer = Uuid::new_v4();
		let node_key = NodeKey {
			nd: peer,
		};
		let nlq_key = NodeLiveQueryKey {
			nd: peer,
			lq: Uuid::new_v4(),
		};
		let seed = Datastore::builder().without_maintenance_tasks().build_with_path(path).await?;
		// Stamp the storage version first: data written to an unstamped datastore
		// reads back as a pre-versioning upgrade, which `init` refuses to start
		// against.
		seed.check_version().await?;
		let txn = seed.transaction(TransactionType::Write).await?;
		txn.set_key(&node_key, &Node::new(peer, Timestamp::default(), true)).await?;
		txn.set_key(
			&nlq_key,
			&NodeLiveQuery {
				ns: NamespaceId(1),
				db: DatabaseId(1),
				tb: "test".to_string().into(),
			},
		)
		.await?;
		txn.commit().await?;
		// Release the storage engine's lock before the caller reopens the path.
		seed.shutdown().await?;
		Ok((node_key, nlq_key))
	}

	/// Leaves a datastore at `path` in the state a version bootstrap that was
	/// interrupted before it stamped `!v` leaves behind: the bootstrap sentinel,
	/// and a node row standing in for the maintenance write that used to reach
	/// storage first.
	///
	/// Written by hand rather than by interrupting a real bootstrap because the
	/// injection registry the core tests use is a test-only build of the
	/// datastore crate; what matters here is that `init` recovers the state, and
	/// the core tests are what pin that the state is the one the bootstrap
	/// actually produces.
	#[cfg(feature = "storage-rocksdb")]
	async fn seed_interrupted_bootstrap(path: &str) -> Result<()> {
		use surrealdb_core::dbs::node::{Node, Timestamp};
		use surrealdb_datastore::TransactionType;
		use surrealdb_datastore::key::schema::BootstrapKey;
		use surrealdb_datastore::version::MajorVersion;
		use uuid::Uuid;

		let seed = Datastore::builder().without_maintenance_tasks().build_with_path(path).await?;
		let txn = seed.transaction(TransactionType::Write).await?;
		txn.set_key(&BootstrapKey {}, &MajorVersion::latest()).await?;
		let peer = Uuid::new_v4();
		txn.set_key(
			&NodeKey {
				nd: peer,
			},
			&Node::new(peer, Timestamp::default(), false),
		)
		.await?;
		txn.commit().await?;
		// Release the storage engine's lock before the caller reopens the path.
		seed.shutdown().await?;
		Ok(())
	}

	/// A server config for `path` with both node-maintenance intervals set to
	/// `interval`, which is what decides whether the node jobs run at all.
	#[cfg(any(feature = "storage-rocksdb", feature = "storage-mem"))]
	fn node_maintenance_config(path: String, interval: Duration) -> Result<Config> {
		use surrealdb_core::options::EngineOptions;

		use crate::ntw::client_ip::ClientIp;

		Ok(Config {
			bind: "127.0.0.1:0".parse()?,
			postgres_bind: None,
			path,
			client_ip: ClientIp::None,
			user: None,
			pass: None,
			crt: None,
			key: None,
			engine: EngineOptions::default()
				.with_node_membership_check_interval(interval)
				.with_node_membership_cleanup_interval(interval),
			no_identification_headers: false,
			allow_origin: Vec::new(),
			durable_session_ttl: None,
		})
	}

	/// Whether the seeded node row and its live-query row are still present,
	/// read in one transaction so the pair is observed at a single snapshot.
	#[cfg(feature = "storage-rocksdb")]
	async fn seeded_rows_present(
		dbs: &Datastore,
		node_key: &NodeKey,
		nlq_key: &NodeLiveQueryKey,
	) -> Result<(bool, bool)> {
		use surrealdb_datastore::TransactionType;

		let txn = dbs.transaction(TransactionType::Read).await?;
		let node = txn.get_key(node_key, None).await?;
		let nlq = txn.get_key(nlq_key, None).await?;
		txn.cancel().await?;
		Ok((node.is_some(), nlq.is_some()))
	}

	/// Startup must never reap archived cluster members on its own thread:
	/// `remove_nodes` opens one write transaction per archived node, so inlining
	/// it would hold readiness for as long as a restarting cluster's residue
	/// takes to clear.
	///
	/// Both intervals are zero, the documented setting that disables the node
	/// jobs. That leaves the schedule without them *and* skips the first pass
	/// `initialise_datastore` spawns, so nothing but the startup steps themselves
	/// could have touched the seeded rows by the time they return.
	#[cfg(feature = "storage-rocksdb")]
	#[test(tokio::test(flavor = "multi_thread"))]
	#[serial]
	async fn init_leaves_archived_nodes_for_the_maintenance_scheduler() -> Result<()> {
		use surrealdb_core::CommunityComposer;
		use surrealdb_observe::NoopObserver;

		let dir = tempfile::tempdir()?;
		let path = format!("rocksdb:{}", dir.path().join("store").display());
		let (node_key, nlq_key) = seed_archived_peer(&path).await?;

		let dbs_opts = TestCli::try_parse_from(["surrealdb"])?.dbs;
		let config = node_maintenance_config(path, Duration::ZERO)?;
		let canceller = CancellationToken::new();
		let (dbs, _recv, _router_state, pending) =
			init(CommunityComposer(), &config, canceller.clone(), Arc::new(NoopObserver), dbs_opts)
				.await?;
		// `init` only builds the datastore; the steps under test are the deferred
		// ones the server runs once its listener is bound.
		initialise_datastore(&dbs, &pending).await?;

		let (node, nlq) = seeded_rows_present(&dbs, &node_key, &nlq_key).await?;
		assert!(node, "startup removed the archived node");
		assert!(nlq, "startup removed the archived node's live query");

		// The cleanup path itself is intact — it is only deferred, not dropped.
		dbs.remove_nodes().await?;
		let (node, nlq) = seeded_rows_present(&dbs, &node_key, &nlq_key).await?;
		assert!(!node, "remove_nodes left the archived node behind");
		assert!(!nlq, "remove_nodes left the archived node's live query behind");

		canceller.cancel();
		dbs.shutdown().await?;
		Ok(())
	}

	/// The first expiry and cleanup pass runs off `initialise_datastore`, after
	/// this node has registered, so a restarting cluster clears dead-member
	/// residue without waiting out an interval and without the reap gating
	/// readiness.
	///
	/// Both intervals are an hour, so the periodic jobs cannot reach their first
	/// deadline inside the test: the only thing that can reap the seeded rows is
	/// the pass `initialise_datastore` spawns.
	#[cfg(feature = "storage-rocksdb")]
	#[test(tokio::test(flavor = "multi_thread"))]
	#[serial]
	async fn init_spawns_the_first_cleanup_pass_after_registration() -> Result<()> {
		use surrealdb_core::CommunityComposer;
		use surrealdb_observe::NoopObserver;

		let dir = tempfile::tempdir()?;
		let path = format!("rocksdb:{}", dir.path().join("store").display());
		let (node_key, nlq_key) = seed_archived_peer(&path).await?;

		let dbs_opts = TestCli::try_parse_from(["surrealdb"])?.dbs;
		let config = node_maintenance_config(path, Duration::from_secs(3600))?;
		let canceller = CancellationToken::new();
		let (dbs, _recv, _router_state, pending) =
			init(CommunityComposer(), &config, canceller.clone(), Arc::new(NoopObserver), dbs_opts)
				.await?;
		// `init` only builds the datastore; the steps under test are the deferred
		// ones the server runs once its listener is bound.
		initialise_datastore(&dbs, &pending).await?;

		// The pass is spawned, so poll for its effect rather than assuming it has
		// already landed when the initialisation returns.
		let deadline = Instant::now() + Duration::from_secs(15);
		let (node, nlq) = loop {
			let present = seeded_rows_present(&dbs, &node_key, &nlq_key).await?;
			if present == (false, false) || Instant::now() >= deadline {
				break present;
			}
			sleep(Duration::from_millis(100)).await;
		};
		assert!(!node, "the first cleanup pass left the archived node behind");
		assert!(!nlq, "the first cleanup pass left the archived node's live query behind");

		canceller.cancel();
		dbs.shutdown().await?;
		Ok(())
	}

	/// `init` builds the datastore and nothing else: every startup step that
	/// opens a transaction waits for `initialise_datastore`, which the server
	/// runs after its listener is bound.
	///
	/// That is what makes a datastore which cannot yet serve a transaction safe
	/// to hand back. An engine that has to join a cluster before it will accept
	/// one can then return its builder immediately and finish joining behind the
	/// readiness gate, rather than holding the bind — and with it `/status` and
	/// `/ready` — until it is a member, which leaves an operator's probes on a
	/// refused connection with no way to tell a starting instance from a dead
	/// one.
	#[cfg(feature = "storage-mem")]
	#[test(tokio::test(flavor = "multi_thread"))]
	async fn init_defers_every_step_that_opens_a_transaction() -> Result<()> {
		use surrealdb_core::CommunityComposer;
		use surrealdb_observe::NoopObserver;

		let dbs_opts = TestCli::try_parse_from(["surrealdb"])?.dbs;
		let config = node_maintenance_config("memory".to_string(), Duration::from_secs(3600))?;
		let canceller = CancellationToken::new();
		let (dbs, _recv, _router_state, pending) =
			init(CommunityComposer(), &config, canceller.clone(), Arc::new(NoopObserver), dbs_opts)
				.await?;

		assert!(
			!dbs.maintenance_tasks_running(),
			"init started the maintenance schedule, whose tasks all write"
		);
		assert!(
			dbs.node_heartbeat_age().await?.is_none(),
			"init registered this node's cluster membership row"
		);

		// The deferred phase is what performs them, so nothing is dropped by
		// being held back.
		initialise_datastore(&dbs, &pending).await?;
		assert!(
			dbs.maintenance_tasks_running(),
			"the deferred initialisation started no maintenance schedule"
		);
		assert!(
			dbs.node_heartbeat_age().await?.is_some(),
			"the deferred initialisation did not register this node"
		);

		canceller.cancel();
		dbs.shutdown().await?;
		Ok(())
	}

	/// The deferred initialisation waits for the backend to report that it can
	/// serve before it opens its first transaction, and that wait is outside the
	/// retry budget the steps share.
	///
	/// A backend that has to join a cluster before it can answer would otherwise
	/// spend that budget on transactions which cannot succeed yet, so a join
	/// slower than `SURREAL_STARTUP_OPERATION_TIMEOUT` would fail a start that
	/// was only ever going to be slow. The budget here is one second and the
	/// backend takes longer than that to report ready, so a wait charged against
	/// it would leave nothing for the steps that follow.
	#[cfg(feature = "storage-mem")]
	#[serial]
	#[test(tokio::test(flavor = "multi_thread"))]
	async fn initialisation_waits_for_the_backend_to_become_serve_ready() -> Result<()> {
		use std::sync::atomic::AtomicBool;

		use common::future::BoxFut;
		use surrealdb_core::CommunityComposer;
		use surrealdb_core::kvs::TransactionBuilderParts;
		use surrealdb_kvs::api::Transactable;
		use surrealdb_kvs::err::Result as KvsResult;
		use surrealdb_kvs::{Metrics, TransactionBuilder, TransactionType};
		use surrealdb_observe::NoopObserver;

		/// The retry budget the startup steps share, set low enough that the
		/// wait under test outlasts it.
		const SHORT_BUDGET: Duration = Duration::from_secs(1);

		/// The community backend behind a gate that holds serve-readiness
		/// closed, recording any transaction opened before it opens.
		struct GatedBuilder {
			inner: Box<dyn TransactionBuilder>,
			gate: CancellationToken,
			ready: Arc<AtomicBool>,
			opened_early: Arc<AtomicBool>,
		}

		impl TransactionBuilder for GatedBuilder {
			fn name(&self) -> &'static str {
				self.inner.name()
			}

			fn new_transaction(
				&self,
				write: TransactionType,
			) -> BoxFut<'_, KvsResult<(Box<dyn Transactable>, bool)>> {
				if !self.ready.load(Ordering::SeqCst) {
					self.opened_early.store(true, Ordering::SeqCst);
				}
				self.inner.new_transaction(write)
			}

			fn shutdown(&self) -> BoxFut<'_, KvsResult<()>> {
				self.inner.shutdown()
			}

			fn register_metrics(&self) -> Option<Metrics> {
				self.inner.register_metrics()
			}

			fn collect_u64_metric(&self, metric: &str) -> Option<u64> {
				self.inner.collect_u64_metric(metric)
			}

			fn wait_until_serve_ready(&self) -> BoxFut<'_, KvsResult<()>> {
				Box::pin(async move {
					self.gate.cancelled().await;
					self.ready.store(true, Ordering::SeqCst);
					Ok(())
				})
			}
		}

		struct GatedComposer {
			gate: CancellationToken,
			ready: Arc<AtomicBool>,
			opened_early: Arc<AtomicBool>,
		}

		impl TransactionBuilderFactory for GatedComposer {
			type RouterState = ();

			async fn new_transaction_builder(
				&self,
				path: &str,
				canceller: CancellationToken,
				config: ConfigMap,
			) -> Result<TransactionBuilderParts<Self::RouterState>> {
				let parts =
					CommunityComposer().new_transaction_builder(path, canceller, config).await?;
				Ok(TransactionBuilderParts::without_router_state(Box::new(GatedBuilder {
					inner: parts.builder,
					gate: self.gate.clone(),
					ready: Arc::clone(&self.ready),
					opened_early: Arc::clone(&self.opened_early),
				})))
			}

			fn path_valid(&self, v: &str) -> Result<String> {
				CommunityComposer().path_valid(v)
			}
		}

		let gate = CancellationToken::new();
		let ready = Arc::new(AtomicBool::new(false));
		let opened_early = Arc::new(AtomicBool::new(false));
		let composer = GatedComposer {
			gate: gate.clone(),
			ready: Arc::clone(&ready),
			opened_early: Arc::clone(&opened_early),
		};

		let _budget = EnvGuard::set(STARTUP_OPERATION_TIMEOUT_ENV, "1s");
		let dbs_opts = TestCli::try_parse_from(["surrealdb"])?.dbs;
		let config = node_maintenance_config("memory".to_string(), Duration::from_secs(3600))?;
		let canceller = CancellationToken::new();
		let (dbs, _recv, _router_state, pending) =
			init(composer, &config, canceller.clone(), Arc::new(NoopObserver), dbs_opts).await?;
		assert!(!ready.load(Ordering::SeqCst), "init waited for the backend to become serve-ready");
		assert_eq!(
			pending.timeout, SHORT_BUDGET,
			"the retry budget this test outlasts was not the one applied"
		);

		let mut deferred = tokio::spawn({
			let dbs = Arc::clone(&dbs);
			async move { initialise_datastore(&dbs, &pending).await }
		});

		// The gate is closed, so the wait cannot complete however long this waits.
		// Outlasting the whole retry budget is what makes the release below prove
		// the wait was not charged against it.
		assert!(
			tokio::time::timeout(SHORT_BUDGET + SHORT_BUDGET / 2, &mut deferred).await.is_err(),
			"the deferred initialisation ran without waiting for serve-readiness"
		);
		assert!(
			!opened_early.load(Ordering::SeqCst),
			"a transaction was opened before the backend reported serve-readiness"
		);

		// More wall-clock has now passed than the whole budget, so a start that
		// completes from here is one whose steps still had theirs.
		gate.cancel();
		deferred.await??;
		assert!(
			!opened_early.load(Ordering::SeqCst),
			"a transaction was opened before the backend reported serve-readiness"
		);

		canceller.cancel();
		dbs.shutdown().await?;
		Ok(())
	}

	/// Startup starts the maintenance schedule itself, because `init` builds the
	/// datastore without one.
	///
	/// Every maintenance task writes — the node-membership refresh blindly
	/// upserts this node's row, every leased job writes its lease — and a write
	/// that reaches storage before `check_version` stamps the version key makes
	/// a datastore this process just created indistinguishable from one written
	/// before that key existed, which `check_version` then refuses as out of
	/// date on this and every later start. Deferring the schedule is what keeps
	/// that ordering structural; starting it here is the other half, and
	/// dropping it would leave a server with no compaction, no reclaim and no
	/// heartbeat while everything else about startup looked healthy.
	#[cfg(feature = "storage-mem")]
	#[test(tokio::test(flavor = "multi_thread"))]
	async fn init_starts_the_maintenance_schedule_after_the_version_gate() -> Result<()> {
		use surrealdb_core::CommunityComposer;
		use surrealdb_datastore::version::MajorVersion;
		use surrealdb_observe::NoopObserver;

		let dbs_opts = TestCli::try_parse_from(["surrealdb"])?.dbs;
		let config = node_maintenance_config("memory".to_string(), Duration::from_secs(3600))?;
		let canceller = CancellationToken::new();
		let (dbs, _recv, _router_state, pending) =
			init(CommunityComposer(), &config, canceller.clone(), Arc::new(NoopObserver), dbs_opts)
				.await?;
		// `init` only builds the datastore; the steps under test are the deferred
		// ones the server runs once its listener is bound.
		initialise_datastore(&dbs, &pending).await?;

		assert!(
			dbs.maintenance_tasks_running(),
			"startup left the datastore with no maintenance tasks"
		);
		// And the gate they were held for did its job: a datastore this start
		// created is stamped at this build, not read as pre-versioning storage.
		assert_eq!(dbs.get_version().await?, (MajorVersion::latest(), false));

		canceller.cancel();
		dbs.shutdown().await?;
		Ok(())
	}
	/// A volume whose first boot was interrupted between claiming the datastore
	/// and stamping it starts, rather than being refused for the life of the
	/// volume.
	///
	/// Run against a persistent backend and through a genuine reopen, because
	/// that is the shape of the failure: the first process wrote and died, and
	/// every later process has only the bytes to go on. Without the sentinel
	/// those bytes read as storage from before `!v` existed, and the version
	/// gate commits `!v = 1` and refuses — permanently, since the refusal is a
	/// permanent error and the stamp it read is now on disk.
	#[cfg(feature = "storage-rocksdb")]
	#[test(tokio::test(flavor = "multi_thread"))]
	#[serial]
	async fn init_completes_a_bootstrap_a_previous_boot_left_unfinished() -> Result<()> {
		use surrealdb_core::CommunityComposer;
		use surrealdb_datastore::version::MajorVersion;
		use surrealdb_observe::NoopObserver;

		let dir = tempfile::tempdir()?;
		let path = format!("rocksdb:{}", dir.path().join("store").display());
		seed_interrupted_bootstrap(&path).await?;

		let dbs_opts = TestCli::try_parse_from(["surrealdb"])?.dbs;
		let config = node_maintenance_config(path, Duration::from_secs(3600))?;
		let canceller = CancellationToken::new();
		let (dbs, _recv, _router_state, pending) =
			init(CommunityComposer(), &config, canceller.clone(), Arc::new(NoopObserver), dbs_opts)
				.await?;
		// `init` only builds the datastore; the steps under test are the deferred
		// ones the server runs once its listener is bound.
		initialise_datastore(&dbs, &pending).await?;

		assert_eq!(
			dbs.get_version().await?,
			(MajorVersion::latest(), false),
			"the interrupted bootstrap was not completed at this build's version"
		);

		canceller.cancel();
		dbs.shutdown().await?;
		Ok(())
	}
}