surrealdb-server 3.3.2

//! The gRPC RPC transport.
//!
//! [`Grpc`] is the third [`RpcProtocol`] implementation, alongside
//! [`Http`](crate::rpc::http::Http) and
//! [`Websocket`](crate::rpc::websocket::Websocket). Every method reaches the
//! database through [`RpcProtocol::execute`], so capability gating, the
//! wall-clock query guard, and the RPC/auth observer events apply here exactly
//! as they do on the other two transports.
//!
//! # Session model
//!
//! gRPC has no connection-scoped state to hang a session on: the SDK's channel
//! reconnects transparently, and requests for one session may arrive on
//! different connections over its lifetime. This transport therefore follows
//! the HTTP model rather than the WebSocket one -- sessions live in one
//! process-wide map, and a request that names a session must prove it may use
//! it.
//!
//! # Security
//!
//! A request that names no session runs on an ephemeral session created for it
//! and dropped afterwards, so it can carry no state between requests and is
//! untargetable by any other caller.
//!
//! A session id is a **capability**: holding one is what authorises using the
//! session behind it. Two rules make that safe to rely on:
//!
//! - Session ids are minted by the server, never chosen by the client. `AttachSession` creates a
//!   session only when the request names none, and answers with a fresh v4 UUID. A caller therefore
//!   cannot squat an id, pre-create one another client will later be handed, or probe for one that
//!   exists -- naming an id that does not exist is refused rather than creating it.
//! - An id is 122 bits of randomness and is never logged or reported to anyone but the client that
//!   was issued it, so learning someone else's amounts to capturing their traffic.
//!
//! This is the same model as a session cookie, and it is what the protocol
//! describes: a client re-attaches to its session after a reconnect, which
//! rules out scoping sessions to a connection the way the WebSocket transport
//! does (a gRPC channel reconnects underneath the client without telling it).
//!
//! On top of that, a request that *does* carry transport-level credentials
//! must present ones matching the session it names -- see
//! [`Grpc::verify_caller_for_session`]. The SDK authenticates by RPC rather
//! than by header, so this does not fire for it; it exists so that a caller
//! which authenticates per request cannot reach a session belonging to a
//! different principal.

use std::collections::{HashMap as StdHashMap, HashSet as StdHashSet, VecDeque};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use dashmap::DashMap;
use futures::{Stream, StreamExt};
use surrealdb_core::channel::{Receiver, bounded};
use surrealdb_core::ctx::CancelHandle;
use surrealdb_core::dbs::{AuthPrincipalSnapshot, QueryStreamJob, Session};
use surrealdb_core::iam::check::check_ns_db;
use surrealdb_core::kvs::Datastore;
use surrealdb_core::rpc::{RpcProtocol, live_query_owner, types_error_from_anyhow};
use surrealdb_datastore::Transaction;
use surrealdb_iam::Auth;
use surrealdb_kvs::TransactionType;
use surrealdb_protocol::method_names;
use surrealdb_protocol::proto::rpc::v1 as rpc;
use surrealdb_protocol::proto::rpc::v1::surreal_db_service_server::SurrealDbService;
use surrealdb_protocol::proto::v1 as proto;
use surrealdb_rpc::capabilities::{MethodTarget, RouteTarget};
use surrealdb_rpc::error::{invalid_params, method_not_allowed, session_exists, session_not_found};
use surrealdb_rpc::{
	DbResult, Method, QUERY_STREAM_BUFFER, QueryResult, QueryStreamItem, QueryType, Token, export,
};
use surrealdb_types::{
	Array, Error as TypesError, ErrorDetails as TypesErrorDetails, HashMap, Notification,
	QueryError, SurrealValue, Value,
};
use tokio::sync::{RwLock, mpsc};
use tonic::{Request, Response, Status};
use uuid::Uuid;
use web_time::Instant;

use crate::cnf::{
	GRPC_MAX_ATTACHED_SESSIONS, GRPC_MAX_MESSAGE_SIZE, GRPC_NOTIFICATION_BUFFER,
	HTTP_MAX_IMPORT_BODY_SIZE, MAX_TRANSACTIONS_PER_SESSION, PKG_NAME, PKG_VERSION,
};
use crate::rpc::RpcState;

/// The size of the chunks a streaming export is broken into.
///
/// Matches the chunk size the SDK sends imports in, so both directions frame
/// their bytes the same way.
const EXPORT_CHUNK_SIZE: usize = surrealdb_protocol::DEFAULT_FILE_CHUNK_SIZE;

/// An export is framed by bytes rather than by records, so its frames are
/// bounded by construction -- provided the fixed chunk still fits the smallest
/// message size an operator can configure. Checked here rather than in a test
/// because both sides are constants: a chunk size raised past the floor should
/// fail the build, not a run.
const _: () = assert!(
	EXPORT_CHUNK_SIZE + (crate::cnf::GRPC_MIN_MESSAGE_SIZE / 256) + (1 << 10)
		<= crate::cnf::GRPC_MIN_MESSAGE_SIZE,
	"an export chunk plus its framing and compression headroom must fit the smallest \
	 configurable message"
);

/// The most records one query batch frame carries, whatever a client asks for.
///
/// A statement's results are split across frames so that no single message
/// grows with the result set: a message past the configured
/// `GRPC_MAX_MESSAGE_SIZE` is refused at both ends, so without this an
/// otherwise valid large `SELECT` is refused outright rather than returning its
/// rows.
///
/// This is the record half of the bound and the reason a frame stays small
/// enough to be worth streaming; [`query_value_budget`] is the byte half and
/// the reason it is never too large to carry. Whichever is reached first ends
/// the frame.
const QUERY_BATCH_RECORDS: usize = 256;

/// Bytes of every message reserved for what the transport adds to a payload.
///
/// Two things sit between the payload and the limit tonic enforces.
///
/// The first is the frame's own fixed fields: its statement and batch indices,
/// its kind, its stats, and the tag and length prefix of each level it nests
/// through. Those are fixed-width scalars measured in tens of bytes, which the
/// kibibyte term covers at any message size.
///
/// The second is compression, and it is proportional rather than fixed. tonic
/// weighs the *compressed* payload against its send limit, and compresses
/// unconditionally once a client has negotiated an encoding -- it does not
/// check whether compressing helped. Data that does not compress therefore
/// grows: measured on incompressible input, gzip at level 6 adds about 0.031%
/// and zstd about 0.003%, both of it stored-block framing. One part in 256 is
/// an order of magnitude more than the worse of the two, and costs 0.4% of a
/// payload that would otherwise be refused for being marginally too large
/// *after* this server had already accepted it.
///
/// A frame carrying an error carries no records, so the two never compete for
/// the same budget.
fn transport_reserve(limit: usize) -> usize {
	(limit / 256) + (1 << 10)
}

/// Bytes charged to a variable-length field on top of its own encoding.
///
/// Such a field -- a record inside a repeated one, an error's message -- is
/// prefixed on the wire by a tag and its length as a varint. The tag is one
/// byte at these field numbers and the varint is at most five for any length
/// that could fit in a message at all, so six is an upper bound rather than an
/// estimate; charging an upper bound is what makes the budget a guarantee
/// rather than a hope.
const WIRE_FIELD_OVERHEAD: usize = 6;

/// The most encoded bytes of variable-sized payload one frame carries: the
/// records of a batch, or the message and details of an error.
///
/// The bound is on the *encoded* size, taken before the frame is handed to
/// tonic, because that is the only measurement both peers agree on. tonic
/// applies the send-side limit to the compressed payload while the receiving
/// client applies its own to the decompressed one, so a frame bounded after
/// compression can still be one the client refuses to decompress -- a failure
/// that surfaces there as a truncated or corrupt stream rather than as the
/// limit it is. Measuring first costs one `encoded_len` walk per record, which
/// the encoder was going to do anyway.
///
/// A single record larger than this cannot be framed under it at any record
/// count, so the statement that produced it fails naming the limit rather than
/// emitting a frame neither end will take.
fn frame_payload_budget() -> usize {
	let limit = *GRPC_MAX_MESSAGE_SIZE;
	limit.saturating_sub(transport_reserve(limit)).max(1)
}

/// Hands back a unary response once it is known to encode within
/// [`frame_payload_budget`], refusing it as a status naming the limit
/// otherwise.
///
/// Bounded against the budget rather than against the advertised size, because
/// the advertised size is what tonic weighs the *compressed* payload against: a
/// response encoding to exactly the limit and failing to compress would be
/// refused by tonic after this had passed it, which is the mid-stream failure
/// this exists to replace.
///
/// Every unary response goes through this, including the ones that carry
/// nothing a caller can influence. Only `Run` can currently return a value
/// large enough to matter, but deciding that per method is a judgement that has
/// to be made again whenever a response gains a field, and getting it wrong is
/// silent: tonic weighs the *compressed* payload against its send limit, so an
/// oversized response that happens to compress goes out and fails inside the
/// client's decompressor, where nothing names a size. `encoded_len` on an empty
/// message is a few instructions, which is a small price for a rule with no
/// exceptions.
///
/// The streaming responses are the deliberate exception, because a stream has
/// no single message to weigh. Each of them bounds its own frames instead:
/// query batches record by record, exports against [`EXPORT_CHUNK_SIZE`], and
/// subscriptions per notification.
fn checked_response<M: surrealdb_types::prost::Message>(message: M) -> Result<M, Status> {
	let size = message.encoded_len();
	let limit = frame_payload_budget();
	if size > limit {
		// `OutOfRange` is the code tonic itself raises for a message past the
		// limit, so the preflight and the enforcement it front-runs classify
		// the same condition the same way -- and it is the code the SDK already
		// maps onto a validation error, which is how the same limit is reported
		// on every other path.
		return Err(Status::out_of_range(format!(
			"The response encodes to {size} bytes, above this server's gRPC message limit of \
			 {limit} bytes. Raise SURREAL_GRPC_MAX_MESSAGE_SIZE on this server."
		)));
	}
	Ok(message)
}

/// Hands back a subscription frame that fits in one message, or the error that
/// replaces one which does not.
///
/// The only frame that can be too large is a notification, whose record a
/// client's own writes decide the size of, and it is a single record with no
/// smaller form -- so what the subscriber gets is the reason instead of the
/// change.
fn bounded_notification(frame: rpc::subscribe_response::Frame) -> rpc::subscribe_response::Frame {
	use surrealdb_types::prost::Message;

	let limit = frame_payload_budget();
	let response = rpc::SubscribeResponse {
		frame: Some(frame),
	};
	let size = response.encoded_len();
	if size <= limit {
		// Restored from the message it was measured in; the field was set
		// immediately above.
		return response.frame.expect("the frame this was built around");
	}
	rpc::subscribe_response::Frame::Error(to_proto_error(&oversized_record(size, limit)))
}

/// What one record costs a frame: its own encoding plus the wire's per-element
/// framing.
fn record_cost(value: &proto::Value) -> usize {
	use surrealdb_types::prost::Message;

	value.encoded_len() + WIRE_FIELD_OVERHEAD
}

/// The failure a record too large for any frame is reported as.
///
/// Named as a size against a limit, and pointing at the setting that moves it,
/// because the alternative a caller has is not to retry but to reconfigure --
/// or to project fewer fields.
fn oversized_record(size: usize, budget: usize) -> TypesError {
	TypesError::validation(
		format!(
			"A single record encodes to {size} bytes, above the {budget} bytes one gRPC message \
			 can carry. Raise SURREAL_GRPC_MAX_MESSAGE_SIZE on this server, or select fewer \
			 fields."
		),
		surrealdb_types::ValidationError::InvalidRequest,
	)
}

/// A registered LIVE query, and the subscription streaming it (once one
/// attaches).
struct LiveQuery {
	/// The session that registered it. Only that session may subscribe.
	session_id: Uuid,
	/// The namespace at registration time, for the active-LQ gauge.
	namespace: Option<String>,
	/// The database at registration time, for the active-LQ gauge.
	database: Option<String>,
	/// The subscription currently streaming this live query's notifications.
	/// At most one at a time -- `GetCapabilities` reports
	/// `multiple_subscribers: false` accordingly.
	subscriber: Option<Subscription>,
}

/// The subscription streaming a live query.
struct Subscription {
	/// Identifies this subscription among the ones a live query has had. A
	/// stream's teardown runs when the client drops it, which may be after it
	/// has already subscribed again, so the teardown compares this before
	/// detaching -- otherwise it would detach its own replacement.
	id: Uuid,
	/// Where this subscription's frames are delivered.
	frames: mpsc::Sender<Result<rpc::SubscribeResponse, Status>>,
}

/// The gRPC RPC handler.
///
/// One instance per server, shared by every gRPC request. See the module docs
/// for how a request's session is resolved and gated.
pub struct Grpc {
	kvs: Arc<Datastore>,
	/// Sessions attached by `AttachSession`, plus the ephemeral ones minted
	/// for requests that name no session.
	sessions: HashMap<Uuid, Arc<RwLock<Session>>>,
	/// The subset of `sessions` created by the transport for a single
	/// request. Tracked so they can never be targeted across requests.
	ephemeral_sessions: HashMap<Uuid, ()>,
	/// Open client-driven transactions, each tagged with the session that
	/// began it so a detach can cancel them.
	transactions: DashMap<Uuid, (Uuid, Arc<Transaction>)>,
	/// How many transactions each session currently holds open, enforcing
	/// [`MAX_TRANSACTIONS_PER_SESSION`].
	///
	/// A slot is reserved before the transaction it admits exists, which is
	/// also what makes this a safe skip-hint for
	/// [`cleanup_txns_filtered`](Self::cleanup_txns_filtered): a session with
	/// no entry here has no transaction open and none on its way.
	transaction_counts: DashMap<Uuid, usize>,
	/// Registered LIVE queries, keyed by live query id.
	live_queries: DashMap<Uuid, LiveQuery>,
	/// How many live queries each session has registered.
	///
	/// A hint that lets a session's sweep skip the registry scan, not a
	/// record of which live queries it holds. Counted before a registration
	/// is published and uncounted after it is removed, so a session with no
	/// entry here has none registered and none on its way, and skipping the
	/// scan cannot miss one.
	///
	/// The scan itself stays over [`Self::live_queries`] deliberately.
	/// Publishing into a registry and a derived set is not atomic, so a sweep
	/// that reads the set can miss a registration that has reached the
	/// registry and not yet the set -- and every per-session sweep is exactly
	/// the thing that must not miss one. A count cannot be missed that way
	/// because it is established first, and being wrong in the other
	/// direction only costs a scan that finds nothing.
	live_query_counts: DashMap<Uuid, usize>,
	metrics_observer: Option<Arc<crate::observe::metrics::MetricsObserver>>,
}

impl Grpc {
	pub fn new(
		kvs: Arc<Datastore>,
		metrics_observer: Option<Arc<crate::observe::metrics::MetricsObserver>>,
	) -> Self {
		Self {
			kvs,
			sessions: HashMap::new(),
			ephemeral_sessions: HashMap::new(),
			transactions: DashMap::new(),
			transaction_counts: DashMap::new(),
			live_queries: DashMap::new(),
			live_query_counts: DashMap::new(),
			metrics_observer,
		}
	}

	/// Registers a session created by the transport for a single request.
	///
	/// The ephemeral marker is inserted *before* the session itself so a
	/// concurrent lookup can never observe the session without also observing
	/// that it is untargetable.
	fn register_ephemeral_session(&self, id: Uuid, session: Arc<RwLock<Session>>) {
		self.ephemeral_sessions.insert(id, ());
		self.sessions.insert(id, session);
	}

	/// How many sessions clients hold open, which is what
	/// [`GRPC_MAX_ATTACHED_SESSIONS`] caps.
	fn attached_session_count(&self) -> usize {
		self.sessions.len().saturating_sub(self.ephemeral_sessions.len())
	}

	/// Drops a request's ephemeral session and anything it left behind.
	async fn remove_ephemeral_session(&self, id: &Uuid) {
		self.sessions.remove(id);
		self.ephemeral_sessions.remove(id);
		self.cleanup_lqs(id).await;
		self.cleanup_txns(id).await;
	}

	/// Whether a caller presenting `caller` may operate on `session_id`.
	///
	/// Holding the session id is the primary authorisation (see the module
	/// docs), so an *unauthenticated* caller -- which is every request from an
	/// SDK, since it authenticates by RPC rather than by header -- passes on
	/// the strength of the id alone.
	///
	/// A session that is not yet bound to a principal is open to whoever holds
	/// its id, exactly as the HTTP transport has it: a session starts
	/// unauthenticated and the caller that attached it must be able to sign in
	/// on it. Once bound, a caller that authenticated at the transport level is
	/// held to the stronger rule the HTTP transport applies: its principal must
	/// match the session's. A caller who presents credentials has told us who
	/// it is, and letting it act on a session belonging to somebody else would
	/// be a privilege escalation whichever of the two identities is the more
	/// privileged.
	///
	/// Ephemeral sessions are refused outright: their ids are minted per
	/// request and must never be reachable from another one.
	async fn verify_caller_for_session(
		&self,
		session_id: &Uuid,
		caller: &Auth,
	) -> Result<(), TypesError> {
		if self.ephemeral_sessions.contains_key(session_id) {
			return Err(session_not_found(*session_id));
		}
		let session_lock = self.get_session(session_id).await?;
		let session = session_lock.read().await;
		// An anonymous caller is the ordinary case, and the id is what
		// authorises it. A session bound to nobody yet has no principal to
		// compare against, so the id is all there is either way.
		if matches!(caller.level(), surrealdb_iam::Level::No)
			|| matches!(session.au.level(), surrealdb_iam::Level::No)
		{
			return Ok(());
		}
		// Roles are deliberately not compared, so a role grant or revocation
		// for the same identity does not lock the legitimate owner out.
		if session.au.id() == caller.id() && session.au.level() == caller.level() {
			Ok(())
		} else {
			Err(session_not_found(*session_id))
		}
	}

	/// Reserves a transaction slot for a session, or reports that the session
	/// already holds the maximum.
	///
	/// Reserving up front (rather than counting the map afterwards) is what
	/// stops two concurrent `begin`s from both observing a below-limit count.
	fn reserve_transaction_slot(&self, session_id: Uuid) -> bool {
		let mut count = self.transaction_counts.entry(session_id).or_insert(0);
		if *count >= *MAX_TRANSACTIONS_PER_SESSION {
			return false;
		}
		*count += 1;
		true
	}

	/// Whether `session_id` is the session that opened `txn`.
	///
	/// Also answers `false` for a transaction that no longer exists, so a
	/// caller cannot tell "not yours" from "not there".
	fn transaction_belongs_to(&self, txn: &Uuid, session_id: Uuid) -> bool {
		self.transactions.get(txn).is_some_and(|entry| entry.value().0 == session_id)
	}

	/// Counts a live query on its way into the registry.
	///
	/// Called before the registration is published, which is what makes an
	/// absent count mean "none registered and none in flight" rather than
	/// merely "none registered yet".
	fn count_live_query(&self, session_id: Uuid) {
		*self.live_query_counts.entry(session_id).or_insert(0) += 1;
	}

	/// Uncounts a live query that has left the registry.
	fn uncount_live_query(&self, session_id: &Uuid) {
		if let Some(mut count) = self.live_query_counts.get_mut(session_id)
			&& *count > 0
		{
			*count -= 1;
		}
		// Drop the guard before pruning: `remove_if` takes the same lock.
		self.live_query_counts.remove_if(session_id, |_, count| *count == 0);
	}

	/// Returns a slot reserved by [`reserve_transaction_slot`](Self::reserve_transaction_slot).
	fn release_transaction_slot(&self, session_id: &Uuid) {
		if let Some(mut count) = self.transaction_counts.get_mut(session_id)
			&& *count > 0
		{
			*count -= 1;
		}
		// Drop the guard before pruning: `remove_if` takes the same lock.
		self.transaction_counts.remove_if(session_id, |_, count| *count == 0);
	}

	/// Delivers a notification to the subscription streaming its live query,
	/// reporting whether this transport owns the live query at all.
	///
	/// Delivery never waits. A subscriber that stops reading fills its queue
	/// and is ended, which is what keeps the memory a subscription can hold to
	/// [`GRPC_NOTIFICATION_BUFFER`]: waiting instead would leave one pending
	/// send per notification, and the dispatcher that calls this keeps
	/// receiving regardless, so nothing would bound how many accumulate.
	/// Ending the subscription rather than dropping the notification is what
	/// tells the subscriber it fell behind -- this transport retains nothing,
	/// so a dropped notification could not be recovered.
	pub(crate) async fn dispatch_notification(&self, notification: &Notification) -> bool {
		let id = notification.id.into_inner();
		// Copy the sender and labels out, and drop the map guard, before
		// awaiting: holding a `DashMap` guard across an await blocks every
		// other user of that shard, including the cleanup that would unblock
		// us.
		let Some((subscriber, namespace, database)) = self.live_queries.get(&id).map(|lq| {
			(
				lq.subscriber.as_ref().map(|s| s.frames.clone()),
				lq.namespace.clone(),
				lq.database.clone(),
			)
		}) else {
			// The live query belongs to another transport.
			return false;
		};
		let Some(subscriber) = subscriber else {
			// The live query has ended even though nothing was streaming it, so
			// the registration goes with it: leaving it would hold the
			// single-subscriber slot for an id that can never produce another
			// notification, and leave the active-LQ gauge counting it forever.
			if notification.action == surrealdb_types::Action::Killed {
				self.forget_live_query(&id);
			}
			// Nothing has subscribed yet. Notifications produced before a
			// subscription attaches are not retained -- this transport reports
			// `LIVE_QUERY_DELIVERY_AT_MOST_ONCE`.
			return true;
		};
		let frame = match notification.action {
			surrealdb_types::Action::Killed => {
				// The live query is gone, so the subscription ends with it.
				rpc::subscribe_response::Frame::End(rpc::SubscribeEnd {
					reason: rpc::SubscribeEndReason::Killed as i32,
					cursor: None,
				})
			}
			// The live query's own WHERE clause or projection is raising an
			// error. A subscriber has to be told -- that is what this action
			// exists for -- and the protocol's only channel for it is the
			// terminal error frame. The registration is left in place so the
			// client can still `KILL` the broken query.
			surrealdb_types::Action::Error => rpc::subscribe_response::Frame::Error(bounded_error(
				proto::SurrealError::new(
					proto::ErrorKind::Query,
					notification.result.clone().into_string().unwrap_or_else(|_| {
						"The live query raised an evaluation error".to_string()
					}),
				),
				frame_payload_budget(),
			)),
			action => match to_proto_notification(notification, action) {
				Ok(notification) => {
					if let Some(observer) = self.metrics_observer.as_ref() {
						observer.record_live_query_notification(
							namespace.as_deref(),
							database.as_deref(),
						);
					}
					rpc::subscribe_response::Frame::Notification(notification)
				}
				// A change this transport cannot render is the subscriber's to
				// know about: silently skipping it would leave a gap in a
				// stream that reports no retention and no resumption.
				Err(error) => rpc::subscribe_response::Frame::Error(to_proto_error(&error)),
			},
		};
		// A notification is one record and cannot be split, so a record too
		// large for a message ends the subscription carrying the reason. That
		// is the same answer this gives a change it cannot render: a stream
		// with no retention and no resumption has no way to skip one and stay
		// truthful about what the subscriber has seen.
		let frame = bounded_notification(frame);
		let mut terminal = !matches!(frame, rpc::subscribe_response::Frame::Notification(_));
		// The last slot is held back for a terminal frame, so a subscriber that
		// has fallen behind can always be told why its stream is ending.
		if !terminal && subscriber.capacity() <= 1 {
			warn!("Ending gRPC subscription to live query {id}: the subscriber is not keeping up");
			subscriber
				.try_send(Err(Status::resource_exhausted(
					"Notifications were produced faster than this subscription read them",
				)))
				.ok();
			terminal = true;
		} else {
			// A closed channel means the client dropped its stream; the
			// subscription guard clears the registration, so nothing to do here.
			subscriber
				.try_send(Ok(rpc::SubscribeResponse {
					frame: Some(frame),
				}))
				.ok();
		}
		if terminal {
			// Dropping the stored sender (and this local clone) closes the
			// channel, which is what ends the client's stream.
			drop(subscriber);
			if notification.action == surrealdb_types::Action::Killed {
				self.forget_live_query(&id);
			} else if let Some(mut entry) = self.live_queries.get_mut(&id) {
				entry.subscriber = None;
			}
		}
		true
	}

	/// Ends a subscription, if one is attached, with the given reason.
	///
	/// The reason is delivered best-effort: every caller drops the
	/// registration, and with it the sender, immediately afterwards, so the
	/// stream ends either way. Waiting for capacity instead would let one
	/// subscriber that has stopped reading block a `KILL`, a detach, or the
	/// shutdown sweep that runs before every other transport is drained.
	fn end_subscription(&self, live_query_id: &Uuid, reason: rpc::SubscribeEndReason) {
		let subscriber = self
			.live_queries
			.get(live_query_id)
			.and_then(|lq| lq.subscriber.as_ref().map(|s| s.frames.clone()));
		if let Some(subscriber) = subscriber {
			let frame = rpc::SubscribeResponse {
				frame: Some(rpc::subscribe_response::Frame::End(rpc::SubscribeEnd {
					reason: reason as i32,
					cursor: None,
				})),
			};
			subscriber.try_send(Ok(frame)).ok();
		}
	}

	/// Ends a live query this transport registered: the same teardown a
	/// subscription's guard performs when it owns the query it is streaming.
	async fn discard_live_query(&self, live_query_id: &Uuid) {
		self.end_subscription(live_query_id, rpc::SubscribeEndReason::Killed);
		self.forget_live_query(live_query_id);
		if let Err(err) = self.kvs.delete_queries(vec![*live_query_id]).await {
			error!("Error discarding live query {live_query_id}: {err}");
		}
	}

	/// Drops a live query's registration and balances the active-LQ gauge.
	///
	/// The only path that unregisters a live query, and so the only place the
	/// session's count is released.
	fn forget_live_query(&self, live_query_id: &Uuid) -> Option<LiveQuery> {
		let (_, entry) = self.live_queries.remove(live_query_id)?;
		self.uncount_live_query(&entry.session_id);
		if let Some(observer) = self.metrics_observer.as_ref() {
			observer.adjust_live_query_active(
				-1,
				entry.namespace.as_deref(),
				entry.database.as_deref(),
			);
		}
		Some(entry)
	}

	/// Cancels every transaction any session left open.
	///
	/// A WebSocket's transactions are reclaimed when its socket closes, but a
	/// gRPC session outlives any one connection, so shutdown is the only point
	/// at which every open transaction is known to be finished with.
	pub(crate) async fn cleanup_all_txns(&self) {
		self.cleanup_txns_filtered(None).await;
	}

	/// Cancels every transaction a session left open.
	async fn cleanup_txns_filtered(&self, session_filter: Option<&Uuid>) {
		// A session that has reserved no slot holds no transaction and has
		// none opening, because the slot is taken before the transaction
		// exists. Skipping on that is what keeps the per-request teardown of
		// an ephemeral session off the size of the whole map -- such a
		// session cannot open a transaction at all, so it never reserves one.
		if let Some(session_id) = session_filter
			&& !self.transaction_counts.contains_key(session_id)
		{
			return;
		}
		// Everything past the hint reads the transaction map itself. A sweep
		// is the one thing that must not miss a transaction, and publishing
		// into a map and a derived index is not atomic: an index would be
		// consulted in the window after the transaction is in the map and
		// before it is in the index, and `reset` -- which sweeps and leaves
		// the session in place -- would return having cancelled nothing.
		let doomed: Vec<Uuid> = self
			.transactions
			.iter()
			.filter(|entry| match session_filter {
				Some(session_id) => &entry.value().0 == session_id,
				None => true,
			})
			.map(|entry| *entry.key())
			.collect();
		for id in doomed {
			if let Some((_, (session_id, tx))) = self.transactions.remove(&id) {
				self.release_transaction_slot(&session_id);
				if let Err(err) = tx.cancel().await {
					warn!("Error cancelling gRPC transaction {id}: {err}");
				}
			}
		}
	}
}

impl RpcProtocol for Grpc {
	fn kvs(&self) -> &Datastore {
		&self.kvs
	}

	fn kvs_arc(&self) -> Arc<Datastore> {
		Arc::clone(&self.kvs)
	}

	fn version_data(&self) -> DbResult {
		DbResult::Other(Value::String(format!("{PKG_NAME}-{}", *PKG_VERSION)))
	}

	fn session_map(&self) -> &HashMap<Uuid, Arc<RwLock<Session>>> {
		&self.sessions
	}

	/// Session enumeration is not offered.
	///
	/// As on HTTP, this transport's sessions are reachable by id from any
	/// connection, so listing them would hand every caller a set of ids to try
	/// the ownership gate against. There is no legitimate per-caller use for
	/// listing other clients' sessions.
	async fn sessions(&self) -> Result<DbResult, TypesError> {
		Err(method_not_allowed(Method::Sessions.to_string()))
	}

	/// Registers a session, subject to [`GRPC_MAX_ATTACHED_SESSIONS`].
	async fn attach(&self, session_id: Uuid) -> Result<DbResult, TypesError> {
		if self.sessions.contains_key(&session_id) {
			return Err(session_exists(session_id));
		}
		// Ephemeral sessions are transient and belong to a request that is
		// still running, so they do not count against the cap on how many
		// sessions clients may hold open.
		if self.attached_session_count() >= *GRPC_MAX_ATTACHED_SESSIONS {
			return Err(method_not_allowed(Method::Attach.to_string()));
		}
		let mut session = Session::default().with_rt(Self::LQ_SUPPORT);
		session.id = Some(session_id);
		self.sessions.insert(session_id, Arc::new(RwLock::new(session)));
		Ok(DbResult::Other(Value::None))
	}

	async fn get_tx(&self, id: Uuid) -> Result<Arc<Transaction>, TypesError> {
		self.transactions
			.get(&id)
			.map(|entry| Arc::clone(&entry.value().1))
			.ok_or_else(|| invalid_params("Transaction not found"))
	}

	// `set_tx` is deliberately left at its default. Nothing calls it -- a
	// transaction only enters the map through `begin`, which is also what
	// tags it with the session that must clean it up -- so an implementation
	// here could only produce an untracked transaction no cleanup path finds.

	const LQ_SUPPORT: bool = true;

	async fn handle_live(
		&self,
		lqid: &Uuid,
		session_id: Uuid,
		namespace: Option<String>,
		database: Option<String>,
	) {
		// Counted before the registration is published, so a sweep either
		// sees the count and searches the registry, or ran entirely before
		// this call and has nothing of this session's to find.
		self.count_live_query(session_id);
		self.live_queries.insert(
			*lqid,
			LiveQuery {
				session_id,
				namespace: namespace.clone(),
				database: database.clone(),
				subscriber: None,
			},
		);
		if let Some(observer) = self.metrics_observer.as_ref() {
			observer.adjust_live_query_active(1, namespace.as_deref(), database.as_deref());
		}
		// Close the register/detach race: `del_session` removes the session
		// from the map *before* sweeping what it owns, so a detach
		// interleaved with the publication above searched a registry this
		// live query had not reached, and would leave it behind with the
		// datastore row it was committed with. Re-checking after publishing
		// means one side always observes the other -- the ordering `begin`
		// relies on for a transaction.
		//
		// A caller only reaches here having found the session (see
		// `register_live_queries`), and a gRPC session id cannot come back
		// once detached -- `attach_session` adopts an existing session and
		// never creates one under a caller's chosen id, and this transport
		// does not persist sessions -- so a session absent now is gone for
		// good rather than replaced by another bearing the same id.
		if !self.sessions.contains_key(&session_id) {
			// Just this registration: anything else under that id belongs to
			// the calls that registered it and answers for itself.
			self.discard_live_query(lqid).await;
			return;
		}
		// The session id is a capability and stays out of the log; the live
		// query id identifies the registration well enough to trace it.
		trace!("Registered live query {lqid} on the gRPC transport");
	}

	async fn cleanup_lqs(&self, session_id: &Uuid) {
		// A session with nothing counted has nothing registered and nothing
		// registering, because the count is taken before the registration is
		// published. Skipping on that is what keeps the per-request teardown
		// of an ephemeral session off the size of the whole map -- such a
		// session is not realtime, so it can register no live query and never
		// counts one.
		if !self.live_query_counts.contains_key(session_id) {
			return;
		}
		// Everything past the hint reads the registry itself, for the reason
		// spelled out on `cleanup_txns_filtered`: a sweep must not miss a
		// registration, and one that consults a derived set misses whatever
		// has reached the registry and not yet the set.
		let doomed: Vec<Uuid> = self
			.live_queries
			.iter()
			.filter(|entry| &entry.value().session_id == session_id)
			.map(|entry| *entry.key())
			.collect();
		for id in doomed {
			self.end_subscription(&id, rpc::SubscribeEndReason::SessionClosed);
			self.forget_live_query(&id);
			if let Err(err) = self.kvs.delete_queries(vec![id]).await {
				error!("Error cleaning up live query {id} on the gRPC transport: {err}");
			}
		}
	}

	async fn cleanup_all_lqs(&self) {
		let doomed: Vec<Uuid> = self.live_queries.iter().map(|entry| *entry.key()).collect();
		for id in doomed {
			self.end_subscription(&id, rpc::SubscribeEndReason::ServerShutdown);
			self.forget_live_query(&id);
			if let Err(err) = self.kvs.delete_queries(vec![id]).await {
				error!("Error cleaning up live query {id} on shutdown: {err}");
			}
		}
	}

	async fn cleanup_txns(&self, session_id: &Uuid) {
		self.cleanup_txns_filtered(Some(session_id)).await;
	}

	async fn begin(&self, _txn: Option<Uuid>, session_id: Uuid) -> Result<DbResult, TypesError> {
		// Reject a `begin` for a session that was never attached, so a client
		// cannot mint an endless stream of fabricated session ids -- each with
		// its own transaction budget and its own counter entry.
		self.get_session(&session_id).await?;
		// A transaction outlives the request that opened it, so it cannot run
		// on an ephemeral session: releasing that session at the end of this
		// request cancels the transaction, and the caller would be holding an
		// id for something that no longer exists.
		if self.ephemeral_sessions.contains_key(&session_id) {
			return Err(invalid_params("Opening a transaction requires an attached session"));
		}
		if !self.reserve_transaction_slot(session_id) {
			return Err(surrealdb_rpc::error::too_many_transactions());
		}
		let tx = match self.kvs.transaction(TransactionType::Write).await {
			Ok(tx) => tx,
			Err(err) => {
				self.release_transaction_slot(&session_id);
				return Err(types_error_from_anyhow(err));
			}
		};
		let id = Uuid::now_v7();
		self.transactions.insert(id, (session_id, Arc::new(tx)));
		// Close the begin/detach race: `del_session` removes the session from
		// the map *before* draining its transactions, so a detach that ran
		// during the await above would have drained the map before this
		// transaction was published. Re-checking after the insert means one
		// side always observes the other.
		if !self.sessions.contains_key(&session_id) {
			self.cleanup_txns_filtered(Some(&session_id)).await;
			return Err(session_not_found(session_id));
		}
		Ok(DbResult::Other(Value::Uuid(surrealdb_types::Uuid::from(id))))
	}

	async fn commit(
		&self,
		txn: Option<Uuid>,
		_session_id: Uuid,
		_params: Array,
	) -> Result<DbResult, TypesError> {
		let txn = txn.ok_or_else(|| invalid_params("Expected a transaction id"))?;
		let Some((_, (session_id, tx))) = self.transactions.remove(&txn) else {
			return Err(invalid_params("Transaction not found"));
		};
		self.release_transaction_slot(&session_id);
		tx.commit().await.map_err(types_error_from_anyhow)?;
		Ok(DbResult::Other(Value::None))
	}

	async fn cancel(
		&self,
		txn: Option<Uuid>,
		_session_id: Uuid,
		_params: Array,
	) -> Result<DbResult, TypesError> {
		let txn = txn.ok_or_else(|| invalid_params("Expected a transaction id"))?;
		let Some((_, (session_id, tx))) = self.transactions.remove(&txn) else {
			return Err(invalid_params("Transaction not found"));
		};
		self.release_transaction_slot(&session_id);
		tx.cancel().await.map_err(types_error_from_anyhow)?;
		Ok(DbResult::Other(Value::None))
	}
}

/// The session a request runs on.
#[derive(Clone, Copy)]
struct ResolvedSession {
	/// The session the method executes against.
	id: Uuid,
	/// The id the client asked for, which session-management methods
	/// (`attach`, `detach`) require and the ownership gate keys on. `None`
	/// when the request named no session and `id` is ephemeral.
	client: Option<Uuid>,
}

/// One gRPC request's view of the server.
///
/// Built per request by the mounting handler so it can carry that request's
/// authenticated [`Session`] -- the principal the ownership gate compares, and
/// the starting state of an ephemeral session.
pub struct GrpcService {
	state: Arc<RpcState>,
	/// The caller's request-level session, as resolved from this request's
	/// headers by the auth middleware.
	caller: Session,
}

impl GrpcService {
	pub fn new(state: Arc<RpcState>, caller: Session) -> Self {
		Self {
			state,
			caller,
		}
	}

	fn rpc(&self) -> &Grpc {
		&self.state.grpc
	}

	fn kvs(&self) -> &Datastore {
		&self.rpc().kvs
	}

	/// Resolves the session a request runs on, applying the ownership gate.
	///
	/// A request naming no session gets an ephemeral one seeded with the
	/// caller's own authentication, which the caller must drop again with
	/// [`release`](Self::release). An ephemeral session is not marked as
	/// supporting realtime, so a `LIVE SELECT` on one is refused rather than
	/// registering a live query that is torn down the moment the request that
	/// created it returns.
	///
	/// `gate` is false only for `AttachSession`, which names the session it is
	/// about to create: there is no session to check ownership of yet, and the
	/// handler has already established that the id was minted here.
	async fn resolve(
		&self,
		context: Option<&rpc::RequestContext>,
		gate: bool,
	) -> Result<ResolvedSession, Status> {
		match context.and_then(|context| context.session.as_ref()) {
			Some(session) => {
				let id = to_uuid(session)?;
				if gate {
					self.rpc()
						.verify_caller_for_session(&id, self.caller.au.as_ref())
						.await
						.map_err(|err| to_status(&err))?;
				}
				Ok(ResolvedSession {
					id,
					client: Some(id),
				})
			}
			None => {
				let id = Uuid::new_v4();
				let mut session = self.caller.clone();
				session.id = Some(id);
				self.rpc().register_ephemeral_session(id, Arc::new(RwLock::new(session)));
				Ok(ResolvedSession {
					id,
					client: None,
				})
			}
		}
	}

	/// Drops a request's ephemeral session, if it minted one.
	async fn release(&self, session: &ResolvedSession) {
		if session.client.is_none() {
			self.rpc().remove_ephemeral_session(&session.id).await;
		}
	}

	/// Runs a method for a request, resolving its session, applying the
	/// request's own timeout ceiling, and dropping any ephemeral session
	/// afterwards.
	async fn execute(
		&self,
		context: Option<&rpc::RequestContext>,
		method: Method,
		params: Array,
	) -> Result<DbResult, Status> {
		// Parse the transaction id before resolving a session: `resolve`
		// registers an ephemeral session, and an early return between the two
		// would strand it in the session map with nothing left to remove it.
		let txn = match context.and_then(|context| context.transaction.as_ref()) {
			Some(txn) => Some(to_uuid(txn)?),
			None => None,
		};
		let session = self.resolve(context, method != Method::Attach).await?;
		// A transaction may only be used by the session that opened it. The
		// map is process-wide, so without this the id alone would be enough
		// for any caller to read, write, commit or cancel inside somebody
		// else's transaction. An unowned id is refused the same way a missing
		// one is, so the answer is not an oracle for which of the two it was.
		if let Some(txn) = txn
			&& !self.rpc().transaction_belongs_to(&txn, session.id)
		{
			self.release(&session).await;
			return Err(to_status(&invalid_params("Transaction not found")));
		}
		let dispatch =
			RpcProtocol::execute(self.rpc(), txn, session.id, session.client, method, params);
		// `RequestContext.timeout` is a ceiling the caller asks for; the
		// server's own `--query-timeout` (applied inside `execute`) still
		// bounds the call independently, so the effective limit is whichever
		// expires first.
		//
		// It applies to `Query` alone, because expiring it drops the dispatch
		// future -- and the RPC and auth observer events fire at the *end* of
		// `RpcProtocol::execute`, so a dropped call is never recorded. Query is
		// the only method here whose wall-clock a caller cannot predict; for
		// the rest, honouring the ceiling would buy a caller nothing it could
		// not get by waiting, while letting it drive `Signin` under a deadline
		// shorter than the password hash and leave no failed-signin audit
		// record behind. Transaction control is doubly exempt: dropping a
		// commit future mid-flight is unsafe.
		let deadline = context
			.and_then(|context| context.timeout.as_ref())
			.and_then(|timeout| Duration::try_from(*timeout).ok())
			.filter(|_| method == Method::Query);
		let result = match deadline {
			Some(deadline) => match tokio::time::timeout(deadline, dispatch).await {
				Ok(result) => result,
				Err(_) => Err(surrealdb_core::rpc::query_timeout_error(deadline)),
			},
			None => dispatch.await,
		};
		self.release(&session).await;
		result.map_err(|err| to_status(&err))
	}

	/// Clones the [`Session`] a request runs against, for the paths that hand
	/// the datastore a session directly (import and export) rather than going
	/// through an RPC method.
	async fn session_for(
		&self,
		context: Option<&rpc::RequestContext>,
		route: RouteTarget,
	) -> Result<Session, Status> {
		if !self.kvs().allows_http_route(&route) {
			warn!("Capabilities denied gRPC route request attempt, target: '{route}'");
			return Err(Status::permission_denied(format!("Route {route} is not allowed")));
		}
		let resolved = self.resolve(context, true).await?;
		let session = match self.rpc().get_session(&resolved.id).await {
			Ok(lock) => lock.read().await.clone(),
			Err(err) => {
				self.release(&resolved).await;
				return Err(to_status(&err));
			}
		};
		self.release(&resolved).await;
		Ok(session)
	}
}

/// The stream type every server-streaming RPC on this service returns.
type ResponseStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send>>;

#[tonic::async_trait]
impl SurrealDbService for GrpcService {
	async fn get_capabilities(
		&self,
		request: Request<rpc::GetCapabilitiesRequest>,
	) -> Result<Response<rpc::GetCapabilitiesResponse>, Status> {
		let request = request.into_inner();
		if let Some(client) = request.client.as_ref() {
			debug!(
				"gRPC client connected: {} {} on {}",
				client.name, client.version, client.platform
			);
		}
		Ok(Response::new(checked_response(rpc::GetCapabilitiesResponse {
			capabilities: Some(self.server_capabilities()),
		})?))
	}

	async fn health(
		&self,
		_request: Request<rpc::HealthRequest>,
	) -> Result<Response<rpc::HealthResponse>, Status> {
		if !self.kvs().allows_http_route(&RouteTarget::Health) {
			return Err(Status::permission_denied("Route health is not allowed"));
		}
		self.kvs().health_check().await.map_err(|err| {
			error!("Health check failed: {err}");
			Status::unavailable("Health check failed")
		})?;
		Ok(Response::new(checked_response(rpc::HealthResponse {})?))
	}

	async fn attach_session(
		&self,
		request: Request<rpc::AttachSessionRequest>,
	) -> Result<Response<rpc::AttachSessionResponse>, Status> {
		let context = request.into_inner().context;
		// Re-attaching to an existing session is what lets a client tell "the
		// server still has my session" from "the server lost it" after a
		// reconnect -- which decides whether its cached namespace, variables
		// and authentication are still in effect.
		if let Some(session) = context.as_ref().and_then(|context| context.session.as_ref()) {
			let id = to_uuid(session)?;
			// A named session is only ever adopted, never created. Creating
			// one here would let a caller choose its own id, and an id a
			// caller can choose is one it can squat, or pre-create so that it
			// already owns the session another client is later handed.
			self.rpc()
				.verify_caller_for_session(&id, self.caller.au.as_ref())
				.await
				.map_err(|err| to_status(&err))?;
			return Ok(Response::new(checked_response(rpc::AttachSessionResponse {
				session: Some(proto::Uuid::from_uuid(id)),
				created: false,
			})?));
		}
		// Creating a session mints its id here, so it is unguessable and
		// unique by construction.
		let id = Uuid::new_v4();
		let context = rpc::RequestContext {
			session: Some(proto::Uuid::from_uuid(id)),
			transaction: None,
			timeout: context.and_then(|context| context.timeout),
		};
		self.execute(Some(&context), Method::Attach, Array::new()).await?;
		Ok(Response::new(checked_response(rpc::AttachSessionResponse {
			session: Some(proto::Uuid::from_uuid(id)),
			created: true,
		})?))
	}

	async fn detach_session(
		&self,
		request: Request<rpc::DetachSessionRequest>,
	) -> Result<Response<rpc::DetachSessionResponse>, Status> {
		let context = request.into_inner().context;
		self.execute(context.as_ref(), Method::Detach, Array::new()).await?;
		Ok(Response::new(checked_response(rpc::DetachSessionResponse {})?))
	}

	async fn reset_session(
		&self,
		request: Request<rpc::ResetSessionRequest>,
	) -> Result<Response<rpc::ResetSessionResponse>, Status> {
		let context = request.into_inner().context;
		self.execute(context.as_ref(), Method::Reset, Array::new()).await?;
		Ok(Response::new(checked_response(rpc::ResetSessionResponse {})?))
	}

	async fn r#use(
		&self,
		request: Request<rpc::UseRequest>,
	) -> Result<Response<rpc::UseResponse>, Status> {
		let request = request.into_inner();
		let params =
			Array::from(vec![from_nullable(request.namespace), from_nullable(request.database)]);
		let result = self.execute(request.context.as_ref(), Method::Use, params).await?;
		// `yuse` answers with the resulting selection so a client can sync its
		// own view; the wire spells "nothing selected" as an empty string.
		let selection = match result {
			DbResult::Other(Value::Object(object)) => object,
			_ => {
				return Err(Status::internal("Use did not report the resulting selection"));
			}
		};
		let field = |key: &str| match selection.get(key) {
			Some(Value::String(value)) => value.clone(),
			_ => String::new(),
		};
		Ok(Response::new(checked_response(rpc::UseResponse {
			namespace: field("namespace"),
			database: field("database"),
		})?))
	}

	async fn set_variable(
		&self,
		request: Request<rpc::SetVariableRequest>,
	) -> Result<Response<rpc::SetVariableResponse>, Status> {
		let request = request.into_inner();
		let value = match request.value {
			Some(value) => from_proto_value(value)?,
			None => Value::None,
		};
		let params = Array::from(vec![Value::String(request.name), value]);
		self.execute(request.context.as_ref(), Method::Set, params).await?;
		Ok(Response::new(checked_response(rpc::SetVariableResponse {})?))
	}

	async fn unset_variable(
		&self,
		request: Request<rpc::UnsetVariableRequest>,
	) -> Result<Response<rpc::UnsetVariableResponse>, Status> {
		let request = request.into_inner();
		let params = Array::from(vec![Value::String(request.name)]);
		self.execute(request.context.as_ref(), Method::Unset, params).await?;
		Ok(Response::new(checked_response(rpc::UnsetVariableResponse {})?))
	}

	async fn signup(
		&self,
		request: Request<rpc::SignupRequest>,
	) -> Result<Response<rpc::SignupResponse>, Status> {
		let request = request.into_inner();
		let credentials = request
			.credentials
			.ok_or_else(|| Status::invalid_argument("Expected signup credentials"))?;
		let params = Array::from(vec![Value::Object(record_credentials(credentials)?)]);
		let result = self.execute(request.context.as_ref(), Method::Signup, params).await?;
		Ok(Response::new(checked_response(rpc::SignupResponse {
			tokens: Some(to_tokens(result)?),
		})?))
	}

	async fn signin(
		&self,
		request: Request<rpc::SigninRequest>,
	) -> Result<Response<rpc::SigninResponse>, Status> {
		let request = request.into_inner();
		let access = request
			.access_method
			.ok_or_else(|| Status::invalid_argument("Expected an access method"))?;
		let params = Array::from(vec![Value::Object(access_credentials(access)?)]);
		let result = self.execute(request.context.as_ref(), Method::Signin, params).await?;
		Ok(Response::new(checked_response(rpc::SigninResponse {
			tokens: Some(to_tokens(result)?),
		})?))
	}

	async fn authenticate(
		&self,
		request: Request<rpc::AuthenticateRequest>,
	) -> Result<Response<rpc::AuthenticateResponse>, Status> {
		let request = request.into_inner();
		let params = Array::from(vec![Value::String(request.token)]);
		let result = self.execute(request.context.as_ref(), Method::Authenticate, params).await?;
		// The session may end up authenticated with a token other than the one
		// presented, so the answer reports what is actually in effect rather
		// than leaving the client to assume its own.
		Ok(Response::new(checked_response(rpc::AuthenticateResponse {
			expires_at: None,
			tokens: to_tokens(result).ok(),
		})?))
	}

	/// Exchanges a refresh token for a fresh pair.
	///
	/// Both halves are needed: `iam::token::refresh` decodes the expired access
	/// token's claims to recover the namespace, database and access method the
	/// new pair is minted for.
	async fn refresh_tokens(
		&self,
		request: Request<rpc::RefreshTokensRequest>,
	) -> Result<Response<rpc::RefreshTokensResponse>, Status> {
		let request = request.into_inner();
		let params = Array::from(vec![token_value(request.access, request.refresh)]);
		let result = self.execute(request.context.as_ref(), Method::Refresh, params).await?;
		Ok(Response::new(checked_response(rpc::RefreshTokensResponse {
			tokens: Some(to_tokens(result)?),
		})?))
	}

	async fn revoke_tokens(
		&self,
		request: Request<rpc::RevokeTokensRequest>,
	) -> Result<Response<rpc::RevokeTokensResponse>, Status> {
		let request = request.into_inner();
		let params = Array::from(vec![token_value(request.access, request.refresh)]);
		self.execute(request.context.as_ref(), Method::Revoke, params).await?;
		Ok(Response::new(checked_response(rpc::RevokeTokensResponse {})?))
	}

	async fn invalidate(
		&self,
		request: Request<rpc::InvalidateRequest>,
	) -> Result<Response<rpc::InvalidateResponse>, Status> {
		let context = request.into_inner().context;
		self.execute(context.as_ref(), Method::Invalidate, Array::new()).await?;
		Ok(Response::new(checked_response(rpc::InvalidateResponse {})?))
	}

	async fn begin_transaction(
		&self,
		request: Request<rpc::BeginTransactionRequest>,
	) -> Result<Response<rpc::BeginTransactionResponse>, Status> {
		let context = request.into_inner().context;
		let result = self.execute(context.as_ref(), Method::Begin, Array::new()).await?;
		let DbResult::Other(Value::Uuid(id)) = result else {
			return Err(Status::internal("Begin did not return a transaction id"));
		};
		Ok(Response::new(checked_response(rpc::BeginTransactionResponse {
			transaction: Some(proto::Uuid::from_uuid(id.into_inner())),
		})?))
	}

	async fn commit_transaction(
		&self,
		request: Request<rpc::CommitTransactionRequest>,
	) -> Result<Response<rpc::CommitTransactionResponse>, Status> {
		let context = request.into_inner().context;
		self.execute(context.as_ref(), Method::Commit, Array::new()).await?;
		Ok(Response::new(checked_response(rpc::CommitTransactionResponse {})?))
	}

	async fn cancel_transaction(
		&self,
		request: Request<rpc::CancelTransactionRequest>,
	) -> Result<Response<rpc::CancelTransactionResponse>, Status> {
		let context = request.into_inner().context;
		self.execute(context.as_ref(), Method::Cancel, Array::new()).await?;
		Ok(Response::new(checked_response(rpc::CancelTransactionResponse {})?))
	}

	type QueryStream = ResponseStream<rpc::QueryResponse>;

	async fn query(
		&self,
		request: Request<rpc::QueryRequest>,
	) -> Result<Response<Self::QueryStream>, Status> {
		let request = request.into_inner();
		// The columnar encoding is defined but not served. Silently answering
		// with row-oriented values would be within the contract, but a client
		// that asked *only* for Arrow has told us it cannot read anything else.
		let arrow_only = !request.accepted_encodings.is_empty()
			&& request
				.accepted_encodings
				.iter()
				.all(|encoding| *encoding == rpc::ResultEncoding::Arrow as i32);
		if arrow_only {
			return Err(Status::unimplemented(
				"This server does not serve columnar (Arrow) query results",
			));
		}
		let batch_records = batch_records(request.max_batch_records);
		self.stream_query(request.context.as_ref(), request.query, request.variables, batch_records)
			.await
	}

	/// Calls a function, or a machine learning model when a version is given.
	///
	/// Reaching `Method::Run` rather than compiling the call to SurrealQL is
	/// what keeps the name a value: it is classified into a function, a model
	/// or a package by the same code every other transport uses, and an
	/// operator's `--deny-rpc run` applies here as it does there.
	async fn run(
		&self,
		request: Request<rpc::RunRequest>,
	) -> Result<Response<rpc::RunResponse>, Status> {
		let request = request.into_inner();
		let args = request
			.args
			.into_iter()
			.map(from_proto_value)
			.collect::<Result<Vec<Value>, Status>>()?;
		// The wire spells "no version" as an empty string; `run` wants it
		// absent, since an empty version names no model.
		let version = if request.version.is_empty() {
			Value::None
		} else {
			Value::String(request.version)
		};
		let params = Array::from(vec![
			Value::String(request.name),
			version,
			Value::Array(Array::from(args)),
		]);
		let result = self.execute(request.context.as_ref(), Method::Run, params).await?;
		let DbResult::Other(value) = result else {
			return Err(Status::internal("Run did not return a value"));
		};
		Ok(Response::new(checked_response(rpc::RunResponse {
			result: Some(to_proto_value(value)?),
		})?))
	}

	/// Ends a live query.
	///
	/// As with `Run`, going through `Method::Kill` is what makes the operation
	/// nameable: the capability gate sees a kill rather than the query it would
	/// otherwise have been compiled into.
	async fn kill(
		&self,
		request: Request<rpc::KillRequest>,
	) -> Result<Response<rpc::KillResponse>, Status> {
		let request = request.into_inner();
		let live_query_id = request
			.live_query_id
			.ok_or_else(|| Status::invalid_argument("Expected a live query id"))?;
		let params = Array::from(vec![Value::Uuid(to_uuid(&live_query_id)?.into())]);
		self.execute(request.context.as_ref(), Method::Kill, params).await?;
		Ok(Response::new(checked_response(rpc::KillResponse {})?))
	}

	type SubscribeStream = ResponseStream<rpc::SubscribeResponse>;

	async fn subscribe(
		&self,
		request: Request<rpc::SubscribeRequest>,
	) -> Result<Response<Self::SubscribeStream>, Status> {
		let request = request.into_inner();
		// A cursor this server cannot honour must be refused, not ignored:
		// resuming from "now" instead would silently skip every change in the
		// gap, which is indistinguishable from there having been none.
		if request.resume_from.is_some() {
			return Err(Status::unimplemented(
				"This server does not retain live query history to resume from",
			));
		}
		let session = self.resolve(request.context.as_ref(), true).await?;
		// A subscription outlives the request that created it, so it cannot
		// run on an ephemeral session -- there would be nothing left to own
		// the live query once this request returned.
		let Some(session_id) = session.client else {
			self.release(&session).await;
			return Err(rejected_before_execution("Subscribing requires an attached session"));
		};
		let (live_query_id, owned) = match request.subscribe_to {
			Some(rpc::subscribe_request::SubscribeTo::LiveQueryId(id)) => (to_uuid(&id)?, false),
			Some(rpc::subscribe_request::SubscribeTo::Query(registration)) => {
				let id = self.register_live_query(request.context.as_ref(), registration).await?;
				(id, true)
			}
			None => {
				return Err(Status::invalid_argument("Expected a live query id or a query"));
			}
		};
		self.attach_subscription(session_id, live_query_id, owned)
	}

	async fn import_surql(
		&self,
		request: Request<tonic::Streaming<rpc::ImportSurqlRequest>>,
	) -> Result<Response<rpc::ImportSurqlResponse>, Status> {
		self.run_import(request.into_inner()).await
	}

	async fn import_ml_model(
		&self,
		request: Request<tonic::Streaming<rpc::ImportMlModelRequest>>,
	) -> Result<Response<rpc::ImportMlModelResponse>, Status> {
		self.run_ml_import(request.into_inner()).await
	}

	type ExportSurqlStream = ResponseStream<rpc::ExportSurqlResponse>;

	async fn export_surql(
		&self,
		request: Request<rpc::ExportSurqlRequest>,
	) -> Result<Response<Self::ExportSurqlStream>, Status> {
		let request = request.into_inner();
		let session = self.session_for(request.context.as_ref(), RouteTarget::Export).await?;
		let config = match request.config {
			Some(config) => from_proto_export_config(config),
			None => export::Config::default(),
		};
		let export = self.start_export(session, config).await?;
		Ok(Response::new(Box::pin(frame_byte_stream(export, |frame| rpc::ExportSurqlResponse {
			frame: Some(match frame {
				ByteFrame::Chunk(chunk) => rpc::export_surql_response::Frame::Chunk(chunk),
				ByteFrame::Trailer(trailer) => rpc::export_surql_response::Frame::Trailer(trailer),
				ByteFrame::Error(error) => rpc::export_surql_response::Frame::Error(error),
			}),
		}))))
	}

	type ExportDirectoryStream = ResponseStream<rpc::ExportDirectoryResponse>;

	/// Directory-format export is not served.
	///
	/// The format is a multi-file layout with its own versioning, compression
	/// and parallelism, and SurrealDB produces only the single SurrealQL
	/// stream that `ExportSurql` serves. `GetCapabilities` does not advertise
	/// `EXPORT_DIRECTORY`.
	async fn export_directory(
		&self,
		_request: Request<rpc::ExportDirectoryRequest>,
	) -> Result<Response<Self::ExportDirectoryStream>, Status> {
		Err(Status::unimplemented(
			"This server does not produce directory-format exports; use ExportSurql",
		))
	}

	type ExportMlModelStream = ResponseStream<rpc::ExportMlModelResponse>;

	async fn export_ml_model(
		&self,
		request: Request<rpc::ExportMlModelRequest>,
	) -> Result<Response<Self::ExportMlModelStream>, Status> {
		let request = request.into_inner();
		let session = self.session_for(request.context.as_ref(), RouteTarget::Ml).await?;
		let export = self.start_ml_export(session, request.name, request.version).await?;
		Ok(Response::new(Box::pin(frame_byte_stream(export, |frame| rpc::ExportMlModelResponse {
			frame: Some(match frame {
				ByteFrame::Chunk(chunk) => rpc::export_ml_model_response::Frame::Chunk(chunk),
				ByteFrame::Trailer(trailer) => {
					rpc::export_ml_model_response::Frame::Trailer(trailer)
				}
				ByteFrame::Error(error) => rpc::export_ml_model_response::Frame::Error(error),
			}),
		}))))
	}
}

/// Route capability targets, paired with the RPC that reaches them, so a
/// denied route is reported by `GetCapabilities` as a denied method.
const ROUTE_METHODS: &[(RouteTarget, &str)] = &[
	(RouteTarget::Health, method_names::HEALTH),
	(RouteTarget::Export, method_names::EXPORT_SURQL),
	(RouteTarget::Import, method_names::IMPORT_SURQL),
	(RouteTarget::Ml, method_names::EXPORT_ML_MODEL),
	(RouteTarget::Ml, method_names::IMPORT_ML_MODEL),
];

/// RPC methods, paired with the RPC that reaches them, for the same reason.
const RPC_METHODS: &[(Method, &str)] = &[
	(Method::Use, method_names::USE),
	(Method::Set, method_names::SET_VARIABLE),
	(Method::Unset, method_names::UNSET_VARIABLE),
	(Method::Signup, method_names::SIGNUP),
	(Method::Signin, method_names::SIGNIN),
	(Method::Authenticate, method_names::AUTHENTICATE),
	(Method::Refresh, method_names::REFRESH_TOKENS),
	(Method::Revoke, method_names::REVOKE_TOKENS),
	(Method::Invalidate, method_names::INVALIDATE),
	(Method::Begin, method_names::BEGIN_TRANSACTION),
	(Method::Commit, method_names::COMMIT_TRANSACTION),
	(Method::Cancel, method_names::CANCEL_TRANSACTION),
	(Method::Query, method_names::QUERY),
	(Method::Run, method_names::RUN),
	(Method::Kill, method_names::KILL),
	(Method::Attach, method_names::ATTACH_SESSION),
	(Method::Detach, method_names::DETACH_SESSION),
	(Method::Reset, method_names::RESET_SESSION),
];

impl GrpcService {
	/// Reports what this build serves, so a client can gate on it rather than
	/// discovering the answer from a failed call.
	fn server_capabilities(&self) -> rpc::ServerCapabilities {
		let mut capabilities = vec![
			"SESSIONS".to_string(),
			"TRANSACTIONS".to_string(),
			"LIVE_QUERIES".to_string(),
			"REFRESH_TOKENS".to_string(),
		];
		if cfg!(feature = "ml") {
			capabilities.push("ML_MODELS".to_string());
		}
		let mut denied = Vec::new();
		for (route, method) in ROUTE_METHODS {
			if !self.kvs().allows_http_route(route) {
				denied.push((*method).to_string());
			}
		}
		for (rpc_method, method) in RPC_METHODS {
			if !self.kvs().allows_rpc_method(&MethodTarget {
				method: *rpc_method,
			}) {
				denied.push((*method).to_string());
			}
		}
		// `Version` names no RPC: denying it withholds the reported build
		// version and says so here, which is the only way a client can tell
		// "denied" from "not reported".
		let version_denied = !self.kvs().allows_rpc_method(&MethodTarget {
			method: Method::Version,
		});
		if version_denied {
			denied.push(method_names::VERSION_PSEUDO_METHOD.to_string());
		}
		rpc::ServerCapabilities {
			server_version: if version_denied {
				String::new()
			} else {
				format!("{PKG_NAME}-{}", *PKG_VERSION)
			},
			low_api_version: None,
			high_api_version: None,
			capabilities,
			denied_methods: denied,
			limits: Some(rpc::Limits {
				max_message_bytes: *GRPC_MAX_MESSAGE_SIZE as u64,
				max_chunk_bytes: EXPORT_CHUNK_SIZE as u64,
				max_query_duration: self
					.kvs()
					.query_timeout()
					.and_then(|timeout| proto::Duration::try_from(timeout).ok()),
				max_batch_records: QUERY_BATCH_RECORDS as u32,
			}),
			live_queries: Some(rpc::LiveQueryCapabilities {
				// Notifications are delivered straight to an attached
				// subscription and never retained, so one produced while
				// nothing is subscribed is lost rather than redelivered.
				delivery: rpc::LiveQueryDelivery::AtMostOnce as i32,
				resumable: false,
				multiple_subscribers: false,
				retention: None,
			}),
			// The codecs the gRPC route enables `accept_compressed` for, most
			// preferred first, so a client can compress an upload without first
			// spending a rejected call to discover whether it may. `identity`
			// is listed last because it is always accepted; naming it is what
			// separates this answer from the silence of a server too old to
			// have the field.
			accepted_message_encodings: crate::ntw::grpc::accepted_message_encodings(),
		}
	}

	/// Runs a query and hands back one result per statement.
	async fn run_query(
		&self,
		context: Option<&rpc::RequestContext>,
		query: String,
		variables: Option<proto::Variables>,
	) -> Result<Vec<QueryResult>, Status> {
		let variables = match variables {
			Some(variables) => from_proto_variables(variables)?,
			None => Value::None,
		};
		let params = Array::from(vec![Value::String(query), variables]);
		match self.execute(context, Method::Query, params).await? {
			DbResult::Query(results) => Ok(results),
			_ => Err(Status::internal("Query did not return statement results")),
		}
	}

	/// Runs a query, framing each statement's results as they are produced.
	///
	/// The stream is built before the query has run: `Begin` goes out first, as
	/// the protocol requires, carrying the statement count parsing already
	/// established. Everything after it is produced lazily, so the first rows
	/// reach the client while the rest of the scan is still happening.
	async fn stream_query(
		&self,
		context: Option<&rpc::RequestContext>,
		query: String,
		variables: Option<proto::Variables>,
		batch_records: usize,
	) -> Result<Response<ResponseStream<rpc::QueryResponse>>, Status> {
		// Parse the transaction id before resolving a session, for the reason
		// `execute` gives: an early return between the two would strand an
		// ephemeral session in the map with nothing left to remove it.
		let txn = match context.and_then(|context| context.transaction.as_ref()) {
			Some(txn) => Some(to_uuid(txn)?),
			None => None,
		};
		let session = self.resolve(context, true).await?;
		if let Some(txn) = txn
			&& !self.rpc().transaction_belongs_to(&txn, session.id)
		{
			self.release(&session).await;
			return Err(to_status(&invalid_params("Transaction not found")));
		}

		let variables = match variables {
			Some(variables) => match from_proto_variables(variables) {
				Ok(variables) => variables,
				Err(status) => {
					self.release(&session).await;
					return Err(status);
				}
			},
			None => Value::None,
		};
		let params = Array::from(vec![Value::String(query), variables]);

		let (items_tx, items_rx) = bounded(QUERY_STREAM_BUFFER);
		// The handle this request cancels with. The frames own it, so it trips
		// when the client stops reading -- see `ReleaseOnDrop`.
		let cancel = CancelHandle::new();
		// A client's own deadline, which `GrpcService::execute` applies for
		// buffered queries and this path would otherwise drop on the floor.
		let requested_timeout = context
			.and_then(|context| context.timeout)
			.and_then(|timeout| Duration::try_from(timeout).ok());
		let (job, principal) = match RpcProtocol::query_stream(
			self.rpc(),
			txn,
			session.id,
			params,
			Some(cancel.clone()),
			items_tx,
		)
		.await
		{
			Ok(job) => job,
			Err(error) => {
				self.release(&session).await;
				return Err(to_status(&error));
			}
		};

		// A failure before execution began -- a parse error, a denied
		// capability, an unknown transaction -- has already been returned
		// above, which is what the protocol asks for: those end the RPC with a
		// transport error and no frames at all.
		let begin = rpc::QueryResponse {
			frame: Some(rpc::query_response::Frame::Begin(rpc::QueryBegin {
				query_id: Some(proto::Uuid::from_uuid(Uuid::new_v4())),
				statement_count: job.statement_count as u32,
			})),
		};

		// The execution is polled only when the client reads, so a deadline
		// awaited alongside it would never fire for a client that stops
		// reading -- and that client would pin an open read snapshot for as
		// long as it held the stream. This watchdog is driven by the runtime
		// instead, so it fires regardless.
		//
		// Tripping the handle alone is not enough: the execution parks on a
		// full `items` channel rather than at a yield point, so the receiver is
		// closed too, which fails that send and lets the executor finalise its
		// transaction on the way out.
		if let Some(deadline) = shortest(requested_timeout, self.rpc().kvs().query_timeout()) {
			let cancel = cancel.clone();
			let items = items_rx.clone();
			tokio::spawn(async move {
				tokio::time::sleep(deadline).await;
				cancel.trip();
				items.close();
			});
		}

		let state = QueryFraming::new(
			Arc::clone(&self.state),
			session,
			principal,
			job,
			items_rx,
			batch_records,
			cancel,
		);
		let frames = futures::stream::unfold(Some(state), |state| async move {
			let mut state = state?;
			state.next().await.map(|frame| (Ok(frame), Some(state)))
		});
		Ok(Response::new(Box::pin(futures::stream::once(async move { Ok(begin) }).chain(frames))))
	}

	/// Registers a live query for a subscription that will own it.
	///
	/// The query runs before it can be checked, and running it is what
	/// registers any `LIVE SELECT` it carries. A registration is only kept
	/// alive by the subscription that owns it, so every one this call does not
	/// hand back is discarded here -- otherwise a query the check rejects
	/// leaves a live query behind with no subscriber and nothing to remove it
	/// before the session ends.
	async fn register_live_query(
		&self,
		context: Option<&rpc::RequestContext>,
		registration: rpc::LiveQueryRegistration,
	) -> Result<Uuid, Status> {
		let results = self.run_query(context, registration.query, registration.variables).await?;
		let registered: Vec<Uuid> = results
			.iter()
			.filter(|result| result.query_type == QueryType::Live)
			.filter_map(|result| match &result.result {
				Ok(Value::Uuid(id)) => Some(id.into_inner()),
				_ => None,
			})
			.collect();
		let outcome = single_live_query(results);
		let kept = outcome.as_ref().ok().copied();
		for id in registered {
			if Some(id) != kept {
				self.rpc().discard_live_query(&id).await;
			}
		}
		outcome
	}

	/// Attaches a subscription to a live query, returning its stream.
	fn attach_subscription(
		&self,
		session_id: Uuid,
		live_query_id: Uuid,
		owned: bool,
	) -> Result<Response<ResponseStream<rpc::SubscribeResponse>>, Status> {
		// One slot beyond the notification buffer, held back so a terminal frame
		// always fits -- a subscriber that stops reading is told why its stream
		// ended rather than seeing it close.
		let (sender, receiver) = mpsc::channel(*GRPC_NOTIFICATION_BUFFER + 1);
		let subscription_id = Uuid::new_v4();
		{
			let Some(mut entry) = self.rpc().live_queries.get_mut(&live_query_id) else {
				return Err(Status::not_found(format!("Live query {live_query_id} not found")));
			};
			// A live query is only visible to the session that registered it:
			// without this check any caller could name another session's live
			// query id and receive its changes.
			if entry.session_id != session_id {
				return Err(Status::not_found(format!("Live query {live_query_id} not found")));
			}
			if entry.subscriber.is_some() {
				return Err(Status::already_exists(format!(
					"Live query {live_query_id} already has a subscriber"
				)));
			}
			entry.subscriber = Some(Subscription {
				id: subscription_id,
				frames: sender,
			});
		}
		let begin = rpc::SubscribeResponse {
			frame: Some(rpc::subscribe_response::Frame::Begin(rpc::SubscribeBegin {
				subscription_id: Some(proto::Uuid::from_uuid(subscription_id)),
				live_query_id: Some(proto::Uuid::from_uuid(live_query_id)),
				cursor: None,
			})),
		};
		let guard = SubscriptionGuard {
			state: Arc::clone(&self.state),
			live_query_id,
			subscription_id,
			owned,
		};
		// The guard travels in the stream's state so that dropping the stream
		// -- which is how a client unsubscribes -- runs the teardown.
		let notifications =
			futures::stream::unfold((receiver, guard), |(mut receiver, guard)| async move {
				receiver.recv().await.map(|item| (item, (receiver, guard)))
			});
		let stream = futures::stream::once(async move { Ok(begin) }).chain(notifications);
		Ok(Response::new(Box::pin(stream)))
	}

	/// Applies a streamed SurrealQL import.
	async fn run_import(
		&self,
		mut frames: tonic::Streaming<rpc::ImportSurqlRequest>,
	) -> Result<Response<rpc::ImportSurqlResponse>, Status> {
		use futures::StreamExt;

		// The context rides on the opening frame, so the session cannot be
		// resolved until it arrives.
		let Some(first) = frames.next().await.transpose()? else {
			return Err(Status::invalid_argument("Import stream carried no frames"));
		};
		let Some(rpc::import_surql_request::Frame::Begin(begin)) = first.frame else {
			return Err(Status::invalid_argument("Import stream must open with a begin frame"));
		};
		let session = self.session_for(begin.context.as_ref(), RouteTarget::Import).await?;
		self.kvs()
			.check(
				&session,
				surrealdb_iam::Action::Edit,
				surrealdb_iam::ResourceKind::Any.on_level(session.au.level().to_owned()),
			)
			.map_err(|err| Status::permission_denied(err.to_string()))?;

		// Frame the request stream as the byte stream the importer consumes,
		// verifying the trailer as the bytes go past so a truncated or
		// corrupted transfer is reported rather than silently half-applied.
		//
		// The transfer is bounded as it goes. `max_decoding_message_size` caps
		// one chunk, not a stream of them, so without this the HTTP route's
		// import limit would be an operator control that `grpc://` ignores.
		let limit = *HTTP_MAX_IMPORT_BODY_SIZE as u64;
		let (bytes_tx, bytes_rx) = surrealdb::channel::bounded::<anyhow::Result<bytes::Bytes>>(1);
		let trailer = tokio::spawn(async move {
			let mut hasher = blake3::Hasher::new();
			let mut streamed: u64 = 0;
			while let Some(frame) = frames.next().await {
				let frame = match frame {
					Ok(frame) => frame,
					Err(status) => {
						bytes_tx.send(Err(anyhow::anyhow!("{}", status.message()))).await.ok();
						return Err(status);
					}
				};
				match frame.frame {
					Some(rpc::import_surql_request::Frame::Chunk(chunk)) => {
						streamed += chunk.data.len() as u64;
						if streamed > limit {
							let refused = Status::resource_exhausted(format!(
								"Import exceeds the {limit} byte limit"
							));
							bytes_tx.send(Err(anyhow::anyhow!("{}", refused.message()))).await.ok();
							return Err(refused);
						}
						hasher.update(&chunk.data);
						if bytes_tx.send(Ok(chunk.data)).await.is_err() {
							// The importer stopped reading, which means it
							// failed; its own error is the one to report.
							return Ok(());
						}
					}
					Some(rpc::import_surql_request::Frame::Trailer(trailer)) => {
						drop(bytes_tx);
						return verify_trailer(&trailer, streamed, hasher.finalize());
					}
					Some(rpc::import_surql_request::Frame::Begin(_)) => {
						return Err(Status::invalid_argument(
							"Import stream carried a second begin frame",
						));
					}
					None => {}
				}
			}
			// A byte stream is complete only once its trailer arrives.
			Err(Status::data_loss(
				"Import stream ended without a trailer; the import may be partially applied",
			))
		});

		let result = self.kvs().import_stream(&session, bytes_rx).await;
		// Take the framing verdict first: an import that failed *because* the
		// transfer was truncated should report the truncation, not the parse
		// error the truncation caused.
		match trailer.await {
			Ok(Ok(())) => {}
			Ok(Err(status)) => return Err(status),
			Err(err) => return Err(Status::internal(format!("Import framing task failed: {err}"))),
		}
		// A whole-stream failure rejects the dump before any of it runs, so it
		// is the request that was wrong and nothing was applied.
		let results = result.map_err(|err| Status::invalid_argument(err.to_string()))?;
		// Per-statement failures are the other case. The import is not
		// transactional — each statement commits on its own and execution
		// carries on past one that fails — so the dump has been executed and
		// the database holds part of it. `failed_precondition` is what says
		// that: the request was well formed, and the state it left behind has
		// to be dealt with before another attempt.
		if let Some(status) = import_failure_status(&results) {
			return Err(status);
		}
		Ok(Response::new(checked_response(rpc::ImportSurqlResponse {})?))
	}

	/// Applies a streamed SurrealML model.
	///
	/// Unlike the SurrealQL import this cannot be applied as it arrives: the
	/// model's own header carries the name and version it is stored under, and
	/// that is only readable once the whole file is in hand. The transfer is
	/// bounded as it goes for the same reason the SurrealQL one is.
	#[cfg(feature = "ml")]
	async fn run_ml_import(
		&self,
		mut frames: tonic::Streaming<rpc::ImportMlModelRequest>,
	) -> Result<Response<rpc::ImportMlModelResponse>, Status> {
		use futures::StreamExt;

		let Some(first) = frames.next().await.transpose()? else {
			return Err(Status::invalid_argument("Import stream carried no frames"));
		};
		let Some(rpc::import_ml_model_request::Frame::Begin(begin)) = first.frame else {
			return Err(Status::invalid_argument("Import stream must open with a begin frame"));
		};
		let session = self.session_for(begin.context.as_ref(), RouteTarget::Ml).await?;
		let (namespace, database) =
			check_ns_db(&session).map_err(|err| rejected_before_execution(err.to_string()))?;
		self.kvs()
			.check(
				&session,
				surrealdb_iam::Action::Edit,
				surrealdb_iam::ResourceKind::Model.on_db(&namespace, &database),
			)
			.map_err(|err| Status::permission_denied(err.to_string()))?;

		let limit = *HTTP_MAX_IMPORT_BODY_SIZE;
		let mut hasher = blake3::Hasher::new();
		let mut model = Vec::new();
		let mut trailer = None;
		while let Some(frame) = frames.next().await {
			match frame?.frame {
				Some(rpc::import_ml_model_request::Frame::Chunk(chunk)) => {
					if model.len() + chunk.data.len() > limit {
						return Err(Status::resource_exhausted(format!(
							"Import exceeds the {limit} byte limit"
						)));
					}
					hasher.update(&chunk.data);
					model.extend_from_slice(&chunk.data);
				}
				Some(rpc::import_ml_model_request::Frame::Trailer(sent)) => {
					trailer = Some(sent);
					break;
				}
				Some(rpc::import_ml_model_request::Frame::Begin(_)) => {
					return Err(Status::invalid_argument(
						"Import stream carried a second begin frame",
					));
				}
				None => {}
			}
		}
		// A byte stream is complete only once its trailer arrives.
		let trailer = trailer.ok_or_else(|| {
			Status::data_loss("Import stream ended without a trailer; the model was not stored")
		})?;
		verify_trailer(&trailer, model.len() as u64, hasher.finalize())?;

		let file = surrealml_core::storage::surml_file::SurMlFile::from_bytes(model)
			.map_err(|err| Status::invalid_argument(format!("Invalid SurrealML file: {err}")))?;
		let (name, version) = (file.header.name.to_string(), file.header.version.to_string());
		// The name and version the model is stored under come from its header,
		// so a file that carries neither has nowhere to go.
		if name.is_empty() || version.is_empty() {
			return Err(Status::invalid_argument("Model name and version must be set"));
		}
		// `Begin` names the model too; refusing a disagreement beats storing it
		// under a name the caller did not ask for.
		if !begin.name.is_empty() && begin.name != name {
			return Err(Status::invalid_argument(format!(
				"The request names model {} but the file carries {name}",
				begin.name
			)));
		}
		if !begin.version.is_empty() && begin.version != version {
			return Err(Status::invalid_argument(format!(
				"The request names version {} but the file carries {version}",
				begin.version
			)));
		}
		let description = file.header.description.to_string();
		self.kvs()
			.put_ml_model(&session, &name, &version, &description, file.to_bytes())
			.await
			.map_err(|err| Status::internal(err.to_string()))?;
		Ok(Response::new(checked_response(rpc::ImportMlModelResponse {})?))
	}

	/// SurrealML support is a build-time feature; without it there is nowhere
	/// to store a model. `GetCapabilities` omits `ML_MODELS` accordingly.
	#[cfg(not(feature = "ml"))]
	async fn run_ml_import(
		&self,
		_frames: tonic::Streaming<rpc::ImportMlModelRequest>,
	) -> Result<Response<rpc::ImportMlModelResponse>, Status> {
		Err(Status::unimplemented("This server was built without SurrealML support"))
	}

	/// Starts a SurrealQL export, handing back the channel its chunks arrive on
	/// and the handle that reports whether it finished.
	async fn start_export(
		&self,
		session: Session,
		config: export::Config,
	) -> Result<Export, Status> {
		let (namespace, database) =
			check_ns_db(&session).map_err(|err| rejected_before_execution(err.to_string()))?;
		self.kvs()
			.check(
				&session,
				surrealdb_iam::Action::View,
				surrealdb_iam::ResourceKind::Any.on_db(&namespace, &database),
			)
			.map_err(|err| Status::permission_denied(err.to_string()))?;
		let (sender, chunks) = surrealdb::channel::bounded(1);
		// A config the export cannot honour is refused here, before any chunk is
		// sent, and it is the caller's to correct — so a structured error keeps its
		// classification instead of arriving as `INTERNAL`, which reads as a server
		// fault worth retrying.
		let task =
			self.kvs().export_with_config(&session, sender, config).await.map_err(|err| {
				match err.downcast_ref::<TypesError>() {
					Some(err) => to_status(err),
					None => Status::internal(err.to_string()),
				}
			})?;
		Ok(Export {
			chunks,
			outcome: tokio::spawn(task),
		})
	}

	/// Starts a SurrealML model export, handing back the channel its chunks
	/// arrive on and the handle that reports whether it finished.
	#[cfg(feature = "ml")]
	async fn start_ml_export(
		&self,
		session: Session,
		name: String,
		version: String,
	) -> Result<Export, Status> {
		let (namespace, database) =
			check_ns_db(&session).map_err(|err| rejected_before_execution(err.to_string()))?;
		self.kvs()
			.check(
				&session,
				surrealdb_iam::Action::View,
				surrealdb_iam::ResourceKind::Model.on_db(&namespace, &database),
			)
			.map_err(|err| Status::permission_denied(err.to_string()))?;
		let info = self
			.kvs()
			.get_db_model(&namespace, &database, &name, &version)
			.await
			.map_err(|err| Status::internal(err.to_string()))?
			.ok_or_else(|| Status::not_found(format!("Model {name} {version} not found")))?;
		let path = format!("ml/{namespace}/{database}/{name}-{version}-{}.surml", info.hash);
		let mut data = surrealdb_core::obs::stream(path)
			.await
			.map_err(|err| Status::internal(format!("Failed to read model file: {err}")))?;
		let (sender, chunks) = surrealdb::channel::bounded(1);
		let outcome = tokio::spawn(async move {
			while let Some(chunk) = data.next().await {
				// A read failure has to reach the stream: stopping quietly
				// would frame a partial model as a complete one.
				let chunk = chunk.map_err(|err| anyhow::anyhow!("{err}"))?;
				if sender.send(chunk.to_vec()).await.is_err() {
					break;
				}
			}
			Ok(())
		});
		Ok(Export {
			chunks,
			outcome,
		})
	}

	/// SurrealML support is a build-time feature; without it there are no
	/// models to export. `GetCapabilities` omits `ML_MODELS` accordingly.
	#[cfg(not(feature = "ml"))]
	async fn start_ml_export(
		&self,
		_session: Session,
		_name: String,
		_version: String,
	) -> Result<Export, Status> {
		Err(Status::unimplemented("This server was built without SurrealML support"))
	}
}

/// An export in flight: the chunks it is producing, and the handle reporting
/// whether it ran to completion.
///
/// Both are needed to frame the stream correctly. The chunk channel closing
/// means the export stopped, but not *why*: a failure part-way through closes
/// it exactly as success does. Reading the outcome is what tells a trailer
/// (this is all of it) from an error frame (this is not).
struct Export {
	chunks: surrealdb::channel::Receiver<Vec<u8>>,
	outcome: tokio::task::JoinHandle<anyhow::Result<()>>,
}

/// A guard that tears a subscription down when its stream is dropped.
///
/// A client unsubscribes by dropping the stream, so this is the only signal
/// that a subscription has ended.
struct SubscriptionGuard {
	state: Arc<RpcState>,
	live_query_id: Uuid,
	/// Which subscription this guard ends. A stream is dropped whenever the
	/// client gets round to it, which may be after it has subscribed again, so
	/// the detach below only fires if this is still the subscription attached.
	subscription_id: Uuid,
	/// Whether the live query was created by this subscription, and so should
	/// be killed along with it.
	owned: bool,
}

impl Drop for SubscriptionGuard {
	fn drop(&mut self) {
		let state = Arc::clone(&self.state);
		let live_query_id = self.live_query_id;
		if !self.owned {
			// The live query outlives its subscribers; just detach.
			if let Some(mut entry) = state.grpc.live_queries.get_mut(&live_query_id)
				&& entry.subscriber.as_ref().is_some_and(|s| s.id == self.subscription_id)
			{
				entry.subscriber = None;
			}
			return;
		}
		// A live query registered by `Subscribe` is bound to that stream, so
		// ending the stream kills it. Teardown is asynchronous, so it runs on
		// its own task.
		tokio::spawn(async move {
			state.grpc.forget_live_query(&live_query_id);
			if let Err(err) = state.grpc.kvs.delete_queries(vec![live_query_id]).await {
				error!("Error killing subscription-owned live query {live_query_id}: {err}");
			}
		});
	}
}

/// The frames a streamed byte transfer is made of.
enum ByteFrame {
	Chunk(rpc::DataChunk),
	Trailer(rpc::DataTrailer),
	Error(proto::SurrealError),
}

/// The sooner of two optional deadlines.
fn shortest(a: Option<Duration>, b: Option<Duration>) -> Option<Duration> {
	match (a, b) {
		(Some(a), Some(b)) => Some(a.min(b)),
		(deadline, None) | (None, deadline) => deadline,
	}
}

/// A streaming execution in flight, as [`QueryStreamJob::run`] hands it over.
type QueryStreamRun = Pin<Box<dyn Future<Output = Result<Vec<QueryResult>, TypesError>> + Send>>;

/// The first batch a statement sends, in records.
///
/// The ramp exists for time-to-first-row: a client waiting on a large `SELECT`
/// sees something after this many records rather than after a full batch. Each
/// subsequent batch doubles up to the negotiated maximum, so the small frames
/// are confined to the start and a long result still costs one frame per
/// [`batch_records`] rows.
const QUERY_FIRST_BATCH_RECORDS: usize = 16;

/// Releases a streaming query's resources when the stream carrying its results
/// is dropped.
///
/// A dropped stream means the client stopped reading, and that is the only
/// signal for it: the framing state is simply never polled again, so nothing
/// after the last `next()` runs. Everything the request owns therefore has to
/// be released from here rather than at the end of the stream.
///
/// Tripping the cancel handle is what stops the query. Waiting for the next
/// send to fail would not be enough: a query inside a phase that emits nothing
/// for a while -- a sort, an aggregate -- would run to completion first.
/// Cancelling this way rather than dropping the execution future is deliberate:
/// the future owns an open transaction, and dropping it would leave that
/// transaction neither committed nor cancelled.
struct ReleaseOnDrop {
	cancel: CancelHandle,
	state: Arc<RpcState>,
	/// The session the request ran on, and so the session the live queries
	/// below were committed under.
	session_id: Uuid,
	/// Whether that session is this request's own and goes with it. A
	/// client-named session outlives the request and is left alone.
	ephemeral: bool,
	/// The live queries the execution has committed that nothing has claimed.
	/// Deleted on the way out -- see [`discard_unregistered_live_queries`].
	unregistered: Vec<Uuid>,
}

impl ReleaseOnDrop {
	fn new(cancel: CancelHandle, state: Arc<RpcState>, session: &ResolvedSession) -> Self {
		Self {
			cancel,
			state,
			session_id: session.id,
			ephemeral: session.client.is_none(),
			unregistered: Vec::new(),
		}
	}

	/// Take responsibility for a live query the execution has committed, until
	/// something registers it.
	fn track_live_query(&mut self, id: Uuid) {
		self.unregistered.push(id);
	}

	/// Give up the live queries collected so far, for the path that settles
	/// them itself.
	fn take_live_queries(&mut self) -> Vec<Uuid> {
		std::mem::take(&mut self.unregistered)
	}

	/// Give up ownership of the ephemeral session, for the path that has
	/// already released it.
	fn released(&mut self) {
		self.ephemeral = false;
	}
}

impl Drop for ReleaseOnDrop {
	fn drop(&mut self) {
		self.cancel.trip();
		let unregistered = std::mem::take(&mut self.unregistered);
		let ephemeral = self.ephemeral.then_some(self.session_id);
		if unregistered.is_empty() && ephemeral.is_none() {
			return;
		}
		// Both steps await, which a `Drop` cannot, so they go to a task.
		// Nothing else holds a handle on this session -- the request that made
		// it is over -- so there is no ordering to preserve against anything
		// outside that task. Inside it the live queries go first: the session
		// carries the namespace and database the orphan metric is attributed
		// to, and removing it would take those labels away.
		let state = Arc::clone(&self.state);
		let session_id = self.session_id;
		tokio::spawn(async move {
			discard_unregistered_live_queries(&state, session_id, unregistered).await;
			if let Some(id) = ephemeral {
				state.grpc.remove_ephemeral_session(&id).await;
			}
		});
	}
}

/// Turns a streaming execution's items into query response frames.
///
/// Holds one statement's rows back only until they fill a batch, so the wire
/// sees them long before the statement -- or the query -- has finished.
struct QueryFraming {
	state: Arc<RpcState>,
	session: ResolvedSession,
	/// The principal the execution runs as, and the one its `LIVE SELECT`s
	/// belong to. The session may be acting as someone else by the time they are
	/// registered.
	principal: AuthPrincipalSnapshot,
	/// The execution, until it has been driven to completion.
	run: Option<QueryStreamRun>,
	items: Receiver<QueryStreamItem>,
	/// The framing decisions, which do not depend on how the items arrive.
	frames: QueryFrames,
	started: Instant,
	/// Set once the terminating frame has been queued, which stops the stream.
	done: bool,
	/// Cancels the query and releases the request's session when this struct is
	/// dropped, i.e. when the client stops reading.
	release: ReleaseOnDrop,
}

/// Turns a streaming execution's items into query response frames.
///
/// Separate from [`QueryFraming`] because none of these decisions depend on how
/// the items arrive: a statement's rows are framed the same whether they were
/// produced one batch at a time or all at once.
struct QueryFrames {
	/// Rows waiting to be framed, and how many batches each statement has sent.
	pending: StdHashMap<u32, PendingStatement>,
	/// Frames produced but not yet yielded, in order.
	queued: VecDeque<rpc::QueryResponse>,
	/// The largest batch this client accepted.
	batch_records: usize,
	/// The most encoded bytes of records one frame may carry, read once at
	/// construction so that a query's framing cannot change under it partway
	/// through a result set.
	value_budget: usize,
	/// Statements already terminated, so a second attempt is ignored.
	///
	/// Two things can end a statement: its own `Finished` item, and a value
	/// this server could not encode. Both have to terminate it -- an encode
	/// failure cannot be deferred to the executor, which does not know about
	/// it -- so whichever happens first wins and the other is dropped. Sending
	/// both would put a frame after a terminal one, which the protocol forbids,
	/// and count the statement twice in `End.result_count`.
	terminated: StdHashSet<u32>,
}

/// One statement's rows on their way to the wire.
#[derive(Default)]
struct PendingStatement {
	values: Vec<proto::Value>,
	/// What each entry of `values` costs on the wire, in the same order.
	///
	/// Kept alongside rather than recomputed at flush time so that draining
	/// part of the pending set can subtract exactly what it took: an
	/// `encoded_len` walk is cheap per record but not free, and a frame that
	/// mis-accounts what it dropped stops being a bound at all.
	costs: Vec<usize>,
	/// What `values` currently costs in total, which is the figure the byte
	/// budget is read against.
	bytes: usize,
	/// How many batches have gone out, which is the frame's `batch_index`.
	batches: u64,
	/// Records to accumulate before flushing, doubling per batch up to the
	/// client's maximum.
	target: usize,
	/// Records sent so far, which the terminal batch reports as the statement's
	/// `records_returned`.
	sent: i64,
	/// Set when the statement's value is a single value rather than a list.
	///
	/// This decides the terminal batch's `kind`, and it cannot be inferred from
	/// the count: a `SELECT` returning one row is still a one-element array,
	/// where `SELECT ONLY` returning one row is that row.
	single: bool,
	/// The uuid a single-value statement produced, kept until the statement
	/// finishes so a `LIVE SELECT`'s id can be read back there -- the point at
	/// which the statement's kind is known.
	live_query: Option<Uuid>,
}

impl PendingStatement {
	/// A statement with nothing sent yet, whose first batch is the small one
	/// the ramp starts from.
	fn new(batch_records: usize) -> Self {
		Self {
			target: QUERY_FIRST_BATCH_RECORDS.min(batch_records).max(1),
			..Default::default()
		}
	}
}

impl QueryFraming {
	fn new(
		state: Arc<RpcState>,
		session: ResolvedSession,
		principal: AuthPrincipalSnapshot,
		job: QueryStreamJob,
		items: Receiver<QueryStreamItem>,
		batch_records: usize,
		cancel: CancelHandle,
	) -> Self {
		let release = ReleaseOnDrop::new(cancel, Arc::clone(&state), &session);
		Self {
			state,
			session,
			principal,
			run: Some(job.run),
			items,
			frames: QueryFrames::new(batch_records),
			started: Instant::now(),
			done: false,
			release,
		}
	}

	/// The next frame, or `None` once the stream is complete.
	async fn next(&mut self) -> Option<rpc::QueryResponse> {
		loop {
			if let Some(frame) = self.frames.pop() {
				return Some(frame);
			}
			if self.done {
				// The session outlives the request that made it only when the
				// client named one; an ephemeral session is this stream's, and
				// nothing else will remove it. Doing it here rather than
				// leaving it to the drop guard keeps it awaited, so a caller
				// that reads the stream to its end sees the session gone.
				if self.session.client.is_none() {
					let state = Arc::clone(&self.state);
					let id = self.session.id;
					self.release.released();
					state.grpc.remove_ephemeral_session(&id).await;
				}
				return None;
			}
			// Drive the execution and the drain together: the channel is
			// bounded, so awaiting either alone would deadlock against the
			// other.
			let run = self.run.as_mut().expect("the execution is driven until it completes");
			tokio::select! {
				item = self.items.recv() => match item {
					Ok(item) => self.absorb(item),
					// The execution dropped its sender, so the only thing left
					// is its own outcome.
					Err(_) => {
						let outcome = self.run.take().expect("still running").await;
						self.finish(outcome).await;
					}
				},
				outcome = run => {
					// The execution finished before its channel drained. Take
					// what is still buffered before terminating.
					self.run = None;
					while let Ok(item) = self.items.try_recv() {
						self.absorb(item);
					}
					self.finish(outcome).await;
				}
			}
		}
	}

	/// Fold one item into the frames it produces, taking responsibility for any
	/// live query it reports.
	///
	/// Until the execution finishes and its live queries are registered, this
	/// request is the only thing that knows those datastore rows exist, so a
	/// stream that never gets that far has to delete them itself.
	fn absorb(&mut self, item: QueryStreamItem) {
		if let Some(id) = self.frames.absorb(item) {
			self.release.track_live_query(id);
		}
	}

	/// Queue the frame that ends the stream.
	async fn finish(&mut self, outcome: Result<Vec<QueryResult>, TypesError>) {
		self.done = true;
		// Whichever way the execution ended, its live queries are settled here,
		// so nothing is left for the drop guard to delete: a registration made
		// below must not then be undone behind the client's back.
		let unregistered = self.release.take_live_queries();
		match outcome {
			Ok(results) => {
				let state = Arc::clone(&self.state);
				let session_id = self.session.id;
				register_live_queries(&state, session_id, &self.principal, &results).await;
				self.frames.end(self.started.elapsed());
			}
			// A failure that belongs to no single statement ends the stream
			// instead of completing it. It carries no results, so there is
			// nothing to register from and every live query the execution
			// committed is left unowned.
			Err(error) => {
				discard_unregistered_live_queries(&self.state, self.session.id, unregistered).await;
				self.frames.error(&error);
			}
		}
	}
}

impl QueryFrames {
	fn new(batch_records: usize) -> Self {
		Self {
			pending: StdHashMap::new(),
			queued: VecDeque::new(),
			batch_records,
			value_budget: frame_payload_budget(),
			terminated: StdHashSet::new(),
		}
	}

	/// The next frame produced so far, in order.
	fn pop(&mut self) -> Option<rpc::QueryResponse> {
		self.queued.pop_front()
	}

	/// Queue the frame that completes the stream successfully.
	fn end(&mut self, elapsed: Duration) {
		self.queued.push_back(rpc::QueryResponse {
			frame: Some(rpc::query_response::Frame::End(rpc::QueryEnd {
				result_count: self.terminated.len() as u32,
				execution_duration: proto::Duration::try_from(elapsed).ok(),
			})),
		});
	}

	/// Queue the frame that ends the stream without completing it.
	fn error(&mut self, error: &TypesError) {
		self.queued.push_back(rpc::QueryResponse {
			frame: Some(rpc::query_response::Frame::Error(to_proto_error(error))),
		});
	}

	/// Fold one item into the frames it produces, reporting the live query the
	/// item's statement committed once that statement has finished.
	fn absorb(&mut self, item: QueryStreamItem) -> Option<Uuid> {
		// A statement this server already terminated -- because a value of its
		// own could not be encoded -- takes nothing further. Its rows would
		// otherwise be framed after its terminal batch, which the protocol
		// forbids and a client reading rows as they arrive would act on.
		if self.terminated.contains(&(item.index() as u32)) {
			return None;
		}
		match item {
			QueryStreamItem::Rows {
				index,
				values,
			} => {
				let index = index as u32;
				let batch_records = self.batch_records;
				self.pending.entry(index).or_insert_with(|| PendingStatement::new(batch_records));
				let encoded = match try_values(values.into_iter()) {
					Ok(encoded) => encoded,
					// A value the wire cannot carry fails only the statement
					// that produced it, exactly as its own error would.
					Err(error) => {
						self.pending.remove(&index);
						self.queue_terminal(
							index,
							rpc::QueryStatementKind::Other,
							Some(to_proto_error(&error)),
							0,
						);
						return None;
					}
				};
				if let Err(error) = self.accumulate(index, encoded) {
					self.pending.remove(&index);
					self.queue_terminal(
						index,
						rpc::QueryStatementKind::Other,
						Some(to_proto_error(&error)),
						0,
					);
				}
				None
			}
			QueryStreamItem::Value {
				index,
				value,
			} => {
				let index = index as u32;
				let batch_records = self.batch_records;
				// Only a `LIVE SELECT`'s id is ever read back from here, so only
				// a uuid is worth keeping: retaining every single statement's
				// value would double the peak memory of the path built to bound
				// it.
				let live_query = match &value {
					Value::Uuid(id) => Some(id.into_inner()),
					_ => None,
				};
				match proto::Value::try_from(value).map_err(types_error_from_anyhow).and_then(
					|value| match record_cost(&value) {
						// A single value is not batched, so there is no record
						// count that would bring it under the budget: it is
						// either carried whole or not at all.
						cost if cost > self.value_budget => {
							Err(oversized_record(cost, self.value_budget))
						}
						cost => Ok((value, cost)),
					},
				) {
					Ok((value, cost)) => {
						// A single value is not a list, so it is not batched:
						// the terminal frame carries it whole, marked `SINGLE`
						// so the client does not rebuild an array around it.
						let entry = self
							.pending
							.entry(index)
							.or_insert_with(|| PendingStatement::new(batch_records));
						entry.values.push(value);
						entry.costs.push(cost);
						entry.bytes += cost;
						entry.single = true;
						entry.live_query = live_query;
					}
					Err(error) => {
						self.pending.remove(&index);
						self.queue_terminal(
							index,
							rpc::QueryStatementKind::Other,
							Some(to_proto_error(&error)),
							0,
						);
					}
				}
				None
			}
			QueryStreamItem::Finished {
				index,
				time,
				query_type,
				error,
			} => {
				let index = index as u32;
				let kind = match query_type {
					QueryType::Live => rpc::QueryStatementKind::Live,
					QueryType::Kill => rpc::QueryStatementKind::Kill,
					_ => rpc::QueryStatementKind::Other,
				};
				// A `LIVE SELECT`'s id rides the stream as its statement's
				// single value, and this is where that statement is known to
				// have produced one -- and to have committed it, since a
				// statement is not finished until its transaction has resolved.
				let live_query = match (query_type, &error) {
					(QueryType::Live, None) => {
						self.pending.get(&index).and_then(|entry| entry.live_query)
					}
					_ => None,
				};
				self.queue_terminal(
					index,
					kind,
					error.map(|e| to_proto_error(&e)),
					time.as_nanos(),
				);
				live_query
			}
		}
	}

	/// Add one statement's rows to its pending set, emitting a frame each time
	/// the set reaches either bound.
	///
	/// Rows are taken one at a time so that both bounds are read between every
	/// pair of them: appending a block and splitting it afterwards would mean
	/// holding the whole block, which is the peak this framing exists to avoid,
	/// and would let a frame pass the byte budget before anything noticed.
	///
	/// Fails on a record whose own encoding is over the budget. No frame can
	/// carry it, so the statement it belongs to ends there -- the same way a
	/// value the wire cannot encode at all ends it -- rather than the whole
	/// query failing or, worse, a frame going out that the client cannot
	/// decode.
	fn accumulate(&mut self, index: u32, encoded: Vec<proto::Value>) -> Result<(), TypesError> {
		for value in encoded {
			let cost = record_cost(&value);
			if cost > self.value_budget {
				return Err(oversized_record(cost, self.value_budget));
			}
			let Some(entry) = self.pending.get_mut(&index) else {
				return Ok(());
			};
			// Close the current frame before the record that would overflow it
			// rather than after, so what goes out is always within budget.
			if !entry.values.is_empty() && entry.bytes + cost > self.value_budget {
				self.flush(index, usize::MAX);
			}
			let Some(entry) = self.pending.get_mut(&index) else {
				return Ok(());
			};
			entry.values.push(value);
			entry.costs.push(cost);
			entry.bytes += cost;
			let target = entry.target;
			if entry.values.len() >= target {
				self.flush(index, target);
			}
		}
		Ok(())
	}

	/// Emit a batch for `index` carrying up to `take` records, leaving the
	/// statement open.
	fn flush(&mut self, index: u32, take: usize) {
		let Some(entry) = self.pending.get_mut(&index) else {
			return;
		};
		let take = take.min(entry.values.len());
		if take == 0 {
			return;
		}
		let values: Vec<proto::Value> = entry.values.drain(..take).collect();
		entry.bytes -= entry.costs.drain(..take).sum::<usize>();
		let batch_index = entry.batches;
		entry.batches += 1;
		entry.sent += values.len() as i64;
		// Ramp toward the client's maximum, so the small frames that make the
		// first row arrive early do not become a per-frame cost on a long result.
		entry.target = (entry.target * 2).min(self.batch_records).max(1);
		self.queued.push_back(rpc::QueryResponse {
			frame: Some(rpc::query_response::Frame::Batch(rpc::QueryBatchFrame {
				query_index: index,
				batch_index,
				// A statement's kind is only known once it finishes, and only
				// the terminal frame needs it: `LIVE` and `KILL` produce a
				// single value rather than rows, so nothing that reaches a
				// non-terminal batch is anything but an ordinary statement.
				statement_kind: rpc::QueryStatementKind::Other as i32,
				kind: rpc::QueryResponseKind::Batched as i32,
				stats: None,
				error: None,
				payload: Some(rpc::query_batch_frame::Payload::Values(rpc::ValueBatch {
					values,
				})),
			})),
		});
	}

	/// Emit the frame that completes a statement, carrying whatever is left.
	fn queue_terminal(
		&mut self,
		index: u32,
		kind: rpc::QueryStatementKind,
		error: Option<proto::SurrealError>,
		elapsed_nanos: u128,
	) {
		if !self.terminated.insert(index) {
			// Already terminated; see `terminated`.
			return;
		}
		let entry = self.pending.remove(&index).unwrap_or_default();
		// A statement whose rows were split reports the count it actually sent;
		// one that failed reports nothing, since its rows are retracted.
		let (payload, records) = if error.is_some() {
			(None, 0)
		} else {
			let records = entry.sent + entry.values.len() as i64;
			(
				Some(rpc::query_batch_frame::Payload::Values(rpc::ValueBatch {
					values: entry.values,
				})),
				records,
			)
		};
		let single = error.is_none() && entry.single;
		let stats = rpc::QueryStats {
			records_returned: records,
			bytes_returned: -1,
			records_scanned: -1,
			bytes_scanned: -1,
			execution_duration: u64::try_from(elapsed_nanos)
				.ok()
				.and_then(|nanos| proto::Duration::try_from(Duration::from_nanos(nanos)).ok()),
		};
		self.queued.push_back(rpc::QueryResponse {
			frame: Some(rpc::query_response::Frame::Batch(rpc::QueryBatchFrame {
				query_index: index,
				batch_index: entry.batches,
				kind: if single {
					rpc::QueryResponseKind::Single as i32
				} else {
					rpc::QueryResponseKind::BatchedFinal as i32
				},
				statement_kind: kind as i32,
				stats: Some(stats),
				error,
				payload,
			})),
		});
	}
}

/// Track the live queries an execution registered.
///
/// A `LIVE SELECT` produces its id as an ordinary result, but the id is only
/// useful once the transport knows which session to deliver that query's
/// notifications to. The buffered path does this over the finished results; a
/// streaming one waits for the same point, since a registration inside a
/// transaction block is not real until the block commits.
///
/// Registrations are only ever added here. A statement that ends a subscription
/// does not report the id it ended, so `dispatch_notification` drops the entry
/// when it sees that id's `Action::Killed` notification instead.
async fn register_live_queries(
	state: &RpcState,
	session_id: Uuid,
	principal: &AuthPrincipalSnapshot,
	results: &[QueryResult],
) {
	if !results.iter().any(|r| r.query_type == QueryType::Live) {
		return;
	}
	// The guard is held across every registration below, taken under the
	// principal the execution ran as. `handle_live` does not touch the session
	// lock, which is what makes holding it safe, and holding it is what stops a
	// registration racing the teardown that would have removed it.
	let session = live_query_owner(state.grpc.session_map(), session_id, principal).await;
	let (namespace, database) = match &session {
		Some(owner) => (owner.ns.clone(), owner.db.clone()),
		// No owner: the session was detached while this query ran, or it is no
		// longer acting as the principal that created these subscriptions --
		// `invalidate` keeps the session and clears the authentication, and its
		// `cleanup_lqs` swept a map these ids were not in yet. Either way
		// registering now would deliver change data under an authorization that
		// has been torn down. Delete the rows the execution committed instead.
		// The `KILL`s need nothing: the statement itself removed their rows,
		// and that cleanup already dropped their registrations.
		None => {
			let orphans = results
				.iter()
				.filter(|result| matches!(result.query_type, QueryType::Live))
				.filter_map(|result| match &result.result {
					Ok(Value::Uuid(id)) => Some(id.into_inner()),
					_ => None,
				})
				.collect();
			discard_unregistered_live_queries(state, session_id, orphans).await;
			return;
		}
	};
	for result in results {
		let Ok(Value::Uuid(id)) = &result.result else {
			continue;
		};
		if result.query_type == QueryType::Live {
			state.grpc.handle_live(id, session_id, namespace.clone(), database.clone()).await;
		}
	}
}

/// Delete the live queries an execution committed that nothing will register.
///
/// A committed `LIVE SELECT` writes a datastore row keyed by its id, and it is
/// the registration in [`register_live_queries`] that makes that row reachable:
/// only a registered live query can be subscribed to, killed, or swept when its
/// session ends. A stream whose registration cannot happen -- abandoned by its
/// client, failed as a whole, or left with no session to register against --
/// would otherwise leave the row behind for the lifetime of the datastore, with
/// the broker producing notifications for it that this transport does not own
/// and no cleanup path can find.
async fn discard_unregistered_live_queries(state: &RpcState, session_id: Uuid, ids: Vec<Uuid>) {
	if ids.is_empty() {
		return;
	}
	let orphaned = ids.len();
	let Err(err) = state.grpc.kvs().delete_queries(ids).await else {
		return;
	};
	error!("Error cleaning up the live queries of an unfinished gRPC streaming query: {err}");
	// The rows survived, so notifications for them will be produced and
	// silently discarded from here on. Count them against the tenant that will
	// be paying for it, so the leak is alertable rather than only greppable.
	if let Some(observer) = state.grpc.metrics_observer.as_ref() {
		let (namespace, database) = match state.grpc.get_session(&session_id).await {
			Ok(lock) => {
				let session = lock.read().await;
				(session.ns.clone(), session.db.clone())
			}
			Err(_) => (None, None),
		};
		for _ in 0..orphaned {
			observer.record_live_query_orphaned(namespace.as_deref(), database.as_deref());
		}
	}
}

/// Frames an export as a byte stream: chunks, then a trailer stating the
/// totals so the receiver can tell a complete transfer from a truncated one --
/// or, if the export failed part-way, an error frame in the trailer's place so
/// a partial transfer is never mistaken for a whole one.
///
/// Frames are sized at [`EXPORT_CHUNK_SIZE`], which is what the server reports
/// as `max_chunk_bytes`. The exporter writes one line per `INSERT` statement,
/// so a line is as wide as its records are: forwarding lines as they arrive
/// would tie frame size to record width and, for wide-enough records, push a
/// frame past the message limit the same handshake advertises. Buffering across
/// short lines and splitting long ones keeps the two agreeing whatever the data
/// looks like.
///
/// The digest and byte count are taken as bytes arrive, not as they are framed,
/// so both describe the export itself and re-framing cannot change them.
fn frame_byte_stream<T, F>(export: Export, wrap: F) -> impl Stream<Item = Result<T, Status>> + Send
where
	F: Fn(ByteFrame) -> T + Send + 'static,
	T: Send + 'static,
{
	/// The state the stream carries between frames. A struct rather than a
	/// stage enum because the hasher is large, and an enum would size every
	/// variant to hold it.
	struct Framing {
		/// `None` once the terminating frame has been emitted, which is what
		/// stops the stream.
		export: Option<Export>,
		/// Received but not yet framed. Bounded by one chunk plus the largest
		/// single line the exporter writes, since a full chunk is emitted as
		/// soon as one is available.
		buffer: bytes::BytesMut,
		hasher: blake3::Hasher,
		streamed: u64,
	}

	futures::stream::unfold(
		(
			Framing {
				export: Some(export),
				buffer: bytes::BytesMut::new(),
				hasher: blake3::Hasher::new(),
				streamed: 0,
			},
			wrap,
		),
		|(mut framing, wrap)| async move {
			loop {
				// A whole chunk is ready: emit exactly that much and keep the
				// rest for the next frame.
				if framing.buffer.len() >= EXPORT_CHUNK_SIZE {
					let data = framing.buffer.split_to(EXPORT_CHUNK_SIZE).freeze();
					let frame = wrap(ByteFrame::Chunk(rpc::DataChunk {
						data,
					}));
					return Some((Ok(frame), (framing, wrap)));
				}
				let export = framing.export.take()?;
				match export.chunks.recv().await {
					Ok(chunk) => {
						framing.hasher.update(&chunk);
						framing.streamed += chunk.len() as u64;
						framing.buffer.extend_from_slice(&chunk);
						framing.export = Some(export);
					}
					// The channel closed, which happens both when the export
					// finished and when it gave up part-way. Whatever is left is
					// short of a full chunk, so it goes out as the last one --
					// and only then does the task's own result say which of the
					// two happened.
					Err(_) if !framing.buffer.is_empty() => {
						let data = std::mem::take(&mut framing.buffer).freeze();
						let frame = wrap(ByteFrame::Chunk(rpc::DataChunk {
							data,
						}));
						// Held so the next poll sees a drained buffer and a
						// closed channel, and terminates.
						framing.export = Some(export);
						return Some((Ok(frame), (framing, wrap)));
					}
					Err(_) => {
						let frame = match export.outcome.await {
							Ok(Ok(())) => ByteFrame::Trailer(rpc::DataTrailer {
								bytes: framing.streamed,
								blake3: framing.hasher.finalize().to_hex().to_string(),
							}),
							Ok(Err(err)) => {
								error!("gRPC export failed: {err}");
								ByteFrame::Error(proto::SurrealError::new(
									proto::ErrorKind::Internal,
									"The export failed part-way through",
								))
							}
							Err(err) => {
								error!("gRPC export task panicked: {err}");
								ByteFrame::Error(proto::SurrealError::new(
									proto::ErrorKind::Internal,
									"The export failed part-way through",
								))
							}
						};
						// `framing.export` stays `None`, ending the stream after
						// this terminating frame.
						return Some((Ok(wrap(frame)), (framing, wrap)));
					}
				}
			}
		},
	)
}

/// The live query id a subscription's own query registered, or why the query
/// was not one a subscription can be built on.
fn single_live_query(mut results: Vec<QueryResult>) -> Result<Uuid, Status> {
	if results.len() != 1 {
		return Err(Status::invalid_argument("Expected exactly one LIVE SELECT statement"));
	}
	let result = results.remove(0);
	if result.query_type != QueryType::Live {
		return Err(Status::invalid_argument("Expected a LIVE SELECT statement"));
	}
	match result.result.map_err(|err| to_status(&err))? {
		Value::Uuid(id) => Ok(id.into_inner()),
		_ => Err(Status::internal("LIVE SELECT did not return a live query id")),
	}
}

/// Checks a byte stream's trailer against what actually arrived.
fn verify_trailer(
	trailer: &rpc::DataTrailer,
	streamed: u64,
	digest: blake3::Hash,
) -> Result<(), Status> {
	if trailer.bytes != streamed {
		return Err(Status::data_loss(format!(
			"Import trailer declared {} bytes but {streamed} arrived",
			trailer.bytes
		)));
	}
	// The checksum is optional; an empty one means the sender did not compute
	// it, which is different from computing a wrong one.
	if !trailer.blake3.is_empty() && !trailer.blake3.eq_ignore_ascii_case(&digest.to_hex()) {
		return Err(Status::data_loss("Import trailer checksum did not match the streamed bytes"));
	}
	Ok(())
}

/// Encodes values for the wire, reporting the first one it cannot carry.
fn try_values(values: impl Iterator<Item = Value>) -> Result<Vec<proto::Value>, TypesError> {
	values.map(|value| proto::Value::try_from(value).map_err(types_error_from_anyhow)).collect()
}

/// How many records to put in one batch, given what the client asked for.
///
/// A request above the server's own bound is clamped rather than refused, which
/// is what `QueryRequest.max_batch_records` requires: a client that asks for
/// more than it may have still gets its results. Zero means the client
/// expressed no preference, so the server chooses. The bound this clamps to is
/// the one reported as `Limits.max_batch_records`, so a client can discover it
/// rather than infer it from the batches it receives.
fn batch_records(requested: u32) -> usize {
	match requested as usize {
		0 => QUERY_BATCH_RECORDS,
		n => n.min(QUERY_BATCH_RECORDS),
	}
}

/// Renders a data-change notification in the shape a subscription streams.
///
/// The lifecycle actions (`Killed`, `Error`) have no `Notification` to render
/// into and are handled by the caller, which is why the action is passed
/// separately rather than read back off the notification.
fn to_proto_notification(
	notification: &Notification,
	action: surrealdb_types::Action,
) -> Result<rpc::Notification, TypesError> {
	let action = match action {
		surrealdb_types::Action::Create => rpc::Action::Created,
		surrealdb_types::Action::Update => rpc::Action::Updated,
		surrealdb_types::Action::Delete => rpc::Action::Deleted,
		// Unreachable: the caller matches these out first.
		surrealdb_types::Action::Killed | surrealdb_types::Action::Error => {
			rpc::Action::Unspecified
		}
	};
	let record_id = match &notification.record {
		Value::RecordId(record) => {
			Some(proto::RecordId::try_from(record.clone()).map_err(types_error_from_anyhow)?)
		}
		_ => None,
	};
	Ok(rpc::Notification {
		live_query_id: Some(proto::Uuid::from_uuid(notification.id.into_inner())),
		action: action as i32,
		record_id,
		value: Some(
			proto::Value::try_from(notification.result.clone()).map_err(types_error_from_anyhow)?,
		),
		cursor: None,
	})
}

/// Renders the tokens an auth method answered with.
///
/// The expiry fields are left unset: the methods answer with the tokens alone,
/// and a client that needs the expiry reads it from the access token's own
/// claims. Reporting a guessed expiry would be worse than reporting none.
fn to_tokens(result: DbResult) -> Result<rpc::Tokens, Status> {
	let DbResult::Other(value) = result else {
		return Err(Status::internal("Authentication did not return a token"));
	};
	// An access method that issues no token at all (record access without a
	// configured JWT) answers with nothing rather than an empty token.
	if matches!(value, Value::None | Value::Null) {
		return Ok(rpc::Tokens::default());
	}
	let (access, refresh) = match Token::from_value(value) {
		Ok(Token::Access(access)) => (access, String::new()),
		Ok(Token::WithRefresh {
			access,
			refresh,
		}) => (access, refresh),
		Err(err) => {
			return Err(Status::internal(format!("Authentication returned no token: {err}")));
		}
	};
	Ok(rpc::Tokens {
		access,
		refresh,
		expires_at: None,
		refresh_expires_at: None,
	})
}

/// Renders an access/refresh pair the way the `refresh` and `revoke` methods
/// expect to receive it.
fn token_value(access: String, refresh: String) -> Value {
	let token = if refresh.is_empty() {
		Token::Access(access)
	} else {
		Token::WithRefresh {
			access,
			refresh,
		}
	};
	token.into_value()
}

/// Flattens an access method back into the credentials object the `signin`
/// method parses, inverting what the SDK's engine builds.
fn access_credentials(access: rpc::AccessMethod) -> Result<surrealdb_types::Object, Status> {
	let method =
		access.method.ok_or_else(|| Status::invalid_argument("Expected an access method"))?;
	let mut object = surrealdb_types::Object::new();
	let mut set = |key: &str, value: String| {
		if !value.is_empty() {
			object.insert(key.to_string(), Value::String(value));
		}
	};
	match method {
		rpc::access_method::Method::User(user) => {
			set("ns", user.namespace);
			set("db", user.database);
			set("user", user.username);
			set("pass", user.password);
			set("ac", user.access);
		}
		rpc::access_method::Method::Bearer(bearer) => {
			set("ns", bearer.namespace);
			set("db", bearer.database);
			set("ac", bearer.access);
			set("key", bearer.key);
		}
		rpc::access_method::Method::Record(record) => {
			return record_credentials(record);
		}
	}
	Ok(object)
}

/// Flattens record credentials into the credentials object, with the access
/// method's own variables alongside the scope fields.
fn record_credentials(record: rpc::RecordCredentials) -> Result<surrealdb_types::Object, Status> {
	let mut object = match record.variables {
		Some(variables) => match from_proto_variables(variables)? {
			Value::Object(object) => object,
			_ => surrealdb_types::Object::new(),
		},
		None => surrealdb_types::Object::new(),
	};
	let mut set = |key: &str, value: String| {
		if !value.is_empty() {
			object.insert(key.to_string(), Value::String(value));
		}
	};
	set("ns", record.namespace);
	set("db", record.database);
	set("ac", record.access);
	Ok(object)
}

/// Maps the wire's export configuration onto the datastore's.
fn from_proto_export_config(config: rpc::ExportConfig) -> export::Config {
	use export::TableConfig;

	let tables = match config.tables.and_then(|tables| tables.selection) {
		Some(rpc::export_config::tables::Selection::All(_)) | None => TableConfig::All,
		Some(rpc::export_config::tables::Selection::None(_)) => TableConfig::None,
		Some(rpc::export_config::tables::Selection::Selected(selected)) => {
			TableConfig::Some(selected.tables)
		}
		Some(rpc::export_config::tables::Selection::Excluded(excluded)) => {
			TableConfig::Exclude(export::ExcludedTables {
				exclude: excluded.tables,
			})
		}
	};
	export::Config {
		// The proto carries no field for the database's own definition, so a
		// request to leave it out is one this transport cannot express and this
		// side cannot detect — an absent field and a deselected one arrive alike.
		// The SDK's own gRPC engine refuses such a request before sending it; a
		// raw client gets the default, which is what it has always got. Read from
		// `default()` rather than written here, so this follows that default and
		// so the compiler points here if the proto gains a field to read instead.
		database_definition: export::Config::default().database_definition,
		users: config.users,
		accesses: config.accesses,
		params: config.params,
		functions: config.functions,
		analyzers: config.analyzers,
		tables,
		versions: config.versions,
		records: config.records,
		sequences: config.sequences,
		apis: config.apis,
		buckets: config.buckets,
		modules: config.modules,
		configs: config.configs,
	}
}

/// Maps a three-state namespace/database selection onto the value the `use`
/// method reads: absent leaves the selection alone, null clears it.
fn from_nullable(value: Option<rpc::NullableString>) -> Value {
	match value.and_then(|value| value.value) {
		Some(rpc::nullable_string::Value::Some(value)) => Value::String(value),
		Some(rpc::nullable_string::Value::Null(_)) => Value::Null,
		None => Value::None,
	}
}

/// Encodes a value for the wire, reporting one it cannot carry.
fn to_proto_value(value: Value) -> Result<proto::Value, Status> {
	proto::Value::try_from(value).map_err(|err| Status::internal(err.to_string()))
}

fn from_proto_value(value: proto::Value) -> Result<Value, Status> {
	Value::try_from(value).map_err(|err| Status::invalid_argument(err.to_string()))
}

fn from_proto_variables(variables: proto::Variables) -> Result<Value, Status> {
	let object = proto::Object {
		items: variables.variables,
	};
	from_proto_value(proto::Value {
		value: Some(proto::value::Value::Object(object)),
	})
}

fn to_uuid(uuid: &proto::Uuid) -> Result<Uuid, Status> {
	uuid.to_uuid().map_err(|err| Status::invalid_argument(err.to_string()))
}

/// Splits typed error details into the `(reason, content)` pair the protocol
/// carries, or `None` when the error has no reason beyond its kind.
///
/// Details nest, and each level serialises as an object tagged the same way:
/// `{ kind, details }`. `ErrorDetails.kind` on the wire is the reason's own
/// name -- `"TransactionConflict"`, `"Auth"`, `"TimedOut"` -- and
/// `ErrorDetails.content` is whatever sits below it, which is either a further
/// tagged object or a leaf payload such as the duration a `TimedOut` query gave
/// up after. Taking both from the serialised form rather than matching every
/// variant is what keeps this from needing a new arm each time a layer gains an
/// error type.
fn to_proto_details(details: &TypesErrorDetails) -> Option<(String, Option<Value>)> {
	let Value::Object(mut outer) = details.clone().into_value() else {
		return None;
	};
	// The outer object is the kind, already carried as `SurrealError.kind`.
	let Some(Value::Object(mut reason)) = outer.remove("details") else {
		return None;
	};
	let Some(Value::String(kind)) = reason.remove("kind") else {
		return None;
	};
	Some((kind, reason.remove("details")))
}

/// Renders an error in the structured shape the protocol carries inside its
/// streams, preserving the kind, message, specific reason, retry hint and
/// cause chain.
///
/// The reason and the retry hint are what a client is required to branch on:
/// the protocol states that `message` is not part of the contract and will be
/// reworded, so an error that arrived carrying only a kind and a message would
/// leave a client with nothing stable to classify it by. Dropping them here
/// would push clients back onto substring matching, which is the failure this
/// part of the schema exists to prevent.
/// Metadata key marking a `FAILED_PRECONDITION` the server raised *before*
/// executing anything.
///
/// That code carries two unlike outcomes. A request refused up front — no
/// namespace selected, no session attached — wrote nothing and is safe to retry
/// once corrected. An import whose statements ran and failed left the database
/// holding part of a dump, and a blind retry applies it twice. A client that
/// branches on the code alone cannot tell them apart.
///
/// The marker rides on the refusals rather than on the partial import so that
/// an older server, which sets neither, is read the safe way round: its
/// unmarked partial-import status still says state may have been written.
pub(crate) const REJECTED_BEFORE_EXECUTION: &str = "surreal-rejected-before-execution";

/// A `FAILED_PRECONDITION` for a request that wrote nothing.
fn rejected_before_execution(message: impl Into<String>) -> Status {
	let mut status = Status::failed_precondition(message.into());
	status.metadata_mut().insert(REJECTED_BEFORE_EXECUTION, "1".parse().expect("a literal 1"));
	status
}

/// The suffix that marks an error message this server had to cut short.
///
/// Present so a reader can tell a trimmed message from one that simply ended,
/// which is the difference between "the error says this" and "the error said
/// more than would fit".
const TRUNCATION_MARKER: &str = " [...truncated]";

fn to_proto_error(error: &TypesError) -> proto::SurrealError {
	bounded_error(proto_error(error), frame_payload_budget())
}

/// Trims an error down to what one frame can carry.
///
/// Nearly every error is a few hundred bytes and passes through untouched. The
/// one that is not is an error carrying user data: `THROW` puts the thrown
/// value into the message, and nothing bounds what a query throws. Emitting
/// that whole would produce a frame the receiving client refuses -- and under
/// compression one this server could not even detect it had produced, since
/// tonic weighs the compressed payload -- which costs the client the error
/// entirely rather than the part of it that did not fit.
///
/// What is shed, in order: the cause chain, then the typed details, then the
/// message itself. The kind survives all three, so an error trimmed to nothing
/// else is still one a client can classify and branch on.
fn bounded_error(mut error: proto::SurrealError, budget: usize) -> proto::SurrealError {
	use surrealdb_types::prost::Message;

	if error.encoded_len() <= budget {
		return error;
	}
	error.cause = None;
	if error.encoded_len() <= budget {
		return error;
	}
	error.details = None;
	if error.encoded_len() <= budget {
		return error;
	}
	// Everything still standing apart from the message is fixed-width, so
	// measuring the error without one says exactly what the message may take.
	let mut fixed = error.clone();
	fixed.message = String::new();
	let room =
		budget.saturating_sub(fixed.encoded_len() + WIRE_FIELD_OVERHEAD + TRUNCATION_MARKER.len());
	error.message = format!("{}{TRUNCATION_MARKER}", truncate_on_boundary(&error.message, room));
	error
}

/// The longest prefix of `text` that is at most `bytes` long and still valid
/// UTF-8.
fn truncate_on_boundary(text: &str, bytes: usize) -> &str {
	if text.len() <= bytes {
		return text;
	}
	let mut end = bytes;
	while end > 0 && !text.is_char_boundary(end) {
		end -= 1;
	}
	&text[..end]
}

fn proto_error(error: &TypesError) -> proto::SurrealError {
	let kind = if error.is_validation() {
		proto::ErrorKind::Validation
	} else if error.is_configuration() {
		proto::ErrorKind::Configuration
	} else if error.is_query() {
		proto::ErrorKind::Query
	} else if error.is_serialization() {
		proto::ErrorKind::Serialization
	} else if error.is_not_allowed() {
		proto::ErrorKind::NotAllowed
	} else if error.is_not_found() {
		proto::ErrorKind::NotFound
	} else if error.is_already_exists() {
		proto::ErrorKind::AlreadyExists
	} else if error.is_connection() {
		proto::ErrorKind::Connection
	} else if error.is_thrown() {
		proto::ErrorKind::Thrown
	} else if error.is_context() {
		proto::ErrorKind::Context
	} else {
		proto::ErrorKind::Internal
	};
	let mut out = proto::SurrealError::new(kind, error.message());
	if let Some((reason, content)) = to_proto_details(error.details()) {
		// A payload that cannot cross the wire costs the leaf detail, not the
		// reason: the reason is the part a client classifies on.
		out =
			out.with_details(reason, content.and_then(|value| proto::Value::try_from(value).ok()));
	}
	// Presence is the whole signal, and a conflict is the one failure the
	// engine reports as transient. No delay is suggested: the client's own
	// backoff is better placed to choose one than a server that cannot see how
	// contended the row is.
	if matches!(error.query_details(), Some(QueryError::TransactionConflict)) {
		out = out.with_retry(None);
	}
	if let Some(cause) = error.cause() {
		out = out.with_cause(proto_error(cause));
	}
	out
}

/// Maps an error onto the gRPC status that terminates a unary call.
///
/// A status carries a code and a message but no structured error, so the
/// reason and the retry hint an [`Error`](TypesError) holds survive only where
/// the protocol has a frame to put them in -- a query batch, or a stream's
/// error frame. A `BEGIN ... COMMIT` block inside a query reports its conflict
/// on the failing statement, and so keeps both; the standalone
/// `CommitTransaction` RPC has an empty response message and can only fail as a
/// status, so a conflict there reaches the client as `ABORTED` and its message.
/// The codes chosen here are the ones the SDK's engine maps back onto the
/// matching error kind.
/// The most bytes of message a gRPC status carries.
///
/// A status travels as the `grpc-message` trailer, so what bounds it is the
/// peer's HTTP/2 header list size rather than the message size -- a far smaller
/// budget, and one the wire can spend three bytes of per byte here, since the
/// trailer is percent-encoded. Nothing truncates it on the way: a header block
/// past that size is refused outright and the call fails as an h2 protocol
/// error, so the caller loses the error rather than the end of it.
///
/// Two kibibytes is several times the longest error this server composes, and
/// leaves the encoded trailer inside the smallest header list a peer is likely
/// to accept.
const STATUS_MESSAGE_BYTES: usize = 2 << 10;

/// Trims an error message to what a status trailer can carry.
///
/// Only a query decides a message this long: `THROW` puts the thrown value in
/// it, and nothing bounds what a query throws.
fn bounded_status_message(message: &str) -> String {
	if message.len() <= STATUS_MESSAGE_BYTES {
		return message.to_string();
	}
	let room = STATUS_MESSAGE_BYTES - TRUNCATION_MARKER.len();
	format!("{}{TRUNCATION_MARKER}", truncate_on_boundary(message, room))
}

fn to_status(error: &TypesError) -> Status {
	let message = bounded_status_message(error.message());
	if error.is_validation() {
		Status::invalid_argument(message)
	} else if error.is_configuration() {
		Status::unimplemented(message)
	} else if error.is_query() || error.is_thrown() {
		Status::aborted(message)
	} else if error.is_not_allowed() {
		Status::permission_denied(message)
	} else if error.is_not_found() {
		Status::not_found(message)
	} else if error.is_already_exists() {
		Status::already_exists(message)
	} else if error.is_connection() {
		Status::unavailable(message)
	} else {
		Status::internal(message)
	}
}

/// The status reporting the statements an import could not apply, or `None`
/// when it applied in full.
///
/// The import path suppresses the results of statements that succeeded, so
/// `results` holds one entry per failure. The verdict is read from the entries
/// themselves rather than from the length, so a successful result arriving here
/// could never be mistaken for a complete import.
///
/// The message is bounded because a gRPC status rides in an HTTP/2 trailer,
/// which a peer may refuse or truncate if it grows: one failure verbatim, a
/// count of the rest, and the fact that the dump is partially applied. The
/// statements are not enumerated — a dump that fails at its first statement
/// fails at every one after it, so the list is as long as the dump.
fn import_failure_status(results: &[QueryResult]) -> Option<Status> {
	/// Bytes of the first failure to quote. The framing below it adds roughly a
	/// hundred more, and `grpc-message` is percent-encoded on the wire, so this
	/// is the quoted text's budget rather than the trailer's.
	const FIRST_FAILURE_BUDGET: usize = 512;

	let mut failures = results.iter().filter_map(|r| r.result.as_ref().err());
	let first = failures.next()?;
	let further = failures.count();

	let first = first.to_string();
	let mut message = truncate_on_boundary(&first, FIRST_FAILURE_BUDGET).to_owned();
	if message.len() < first.len() {
		message.push_str(TRUNCATION_MARKER);
	}
	if further > 0 {
		message.push_str(&format!(" (and {further} further statements failed)"));
	}
	message.push_str(
		"; the import is not transactional, so the statements before this one have been applied",
	);
	Some(Status::failed_precondition(message))
}

#[cfg(test)]
mod tests {
	use surrealdb_rpc::capabilities::Capabilities;
	use surrealdb_types::{AuthError, NotAllowedError, Object, SerializationError};

	use super::*;

	async fn service() -> GrpcService {
		service_for(Session::default()).await
	}

	/// A service whose requests arrive as `caller` -- the principal the auth
	/// middleware resolved from the request's own headers.
	async fn service_for(caller: Session) -> GrpcService {
		let datastore = Datastore::builder()
			.with_capabilities(Capabilities::all())
			.build_with_path("memory")
			.await
			.expect("datastore");
		GrpcService::new(Arc::new(RpcState::new(datastore)), caller)
	}

	/// Registers a live query the way the transport does, rather than by
	/// inserting into the registry: `handle_live` also counts it against its
	/// session, and that count is what decides whether the session's teardown
	/// searches the registry at all.
	///
	/// The session is attached first because a registration only survives one
	/// that exists -- `handle_live` discards a live query whose session has
	/// gone, which is what closes the register/detach race.
	async fn register_live_query(service: &GrpcService, session_id: Uuid) -> Uuid {
		if !service.rpc().sessions.contains_key(&session_id) {
			service.rpc().attach(session_id).await.expect("attach");
		}
		let id = Uuid::new_v4();
		service.rpc().handle_live(&id, session_id, None, None).await;
		id
	}

	/// The live queries the registry holds for a session, sorted so a test can
	/// compare them. This is what the session's teardown sweeps, so it is what
	/// a test asserting reclamation has to read.
	fn registered_live_queries(service: &GrpcService, session_id: Uuid) -> Vec<Uuid> {
		let mut ids: Vec<Uuid> = service
			.rpc()
			.live_queries
			.iter()
			.filter(|entry| entry.value().session_id == session_id)
			.map(|entry| *entry.key())
			.collect();
		ids.sort();
		ids
	}

	/// The session's live-query count, or `None` when it holds no entry --
	/// which is the state that makes its teardown skip the registry scan.
	fn live_query_count(service: &GrpcService, session_id: Uuid) -> Option<usize> {
		service.rpc().live_query_counts.get(&session_id).map(|entry| *entry.value())
	}

	/// Opens a transaction on an attached session and hands back its id.
	async fn begin_txn(service: &GrpcService, session_id: Uuid) -> Uuid {
		let DbResult::Other(Value::Uuid(txn)) =
			service.rpc().begin(None, session_id).await.expect("begin")
		else {
			panic!("begin should answer with a transaction id");
		};
		txn.into_inner()
	}

	/// A session starts life bound to nobody, so whoever holds its id may drive
	/// it -- including the caller that attached it while presenting
	/// credentials of its own. Comparing principals before the session has one
	/// would lock that caller out of the session it just created, with no way
	/// to sign in on it.
	#[tokio::test]
	async fn a_credentialed_caller_may_use_the_session_it_attached() {
		let service = service_for(Session::owner()).await;
		let session_id = Uuid::new_v4();
		service.rpc().attach(session_id).await.expect("attach");

		service
			.rpc()
			.verify_caller_for_session(&session_id, service.caller.au.as_ref())
			.await
			.expect("the caller that attached the session may use it");
	}

	/// Once a session is bound to a principal, a caller that authenticated at
	/// the transport level must be that principal. Anything else is a
	/// privilege change in one direction or the other.
	#[tokio::test]
	async fn a_credentialed_caller_may_not_use_another_principals_session() {
		let service = service_for(Session::owner()).await;
		let session_id = Uuid::new_v4();
		service.rpc().attach(session_id).await.expect("attach");
		// Bind the session to a different principal than the caller's.
		{
			let session = service.rpc().get_session(&session_id).await.expect("session");
			let mut session = session.write().await;
			session.au = Arc::new(Auth::for_ns(surrealdb_iam::Role::Owner, "test-ns"));
		}

		let refused = service
			.rpc()
			.verify_caller_for_session(&session_id, service.caller.au.as_ref())
			.await
			.expect_err("a different principal must be refused");
		assert!(refused.is_not_found(), "expected a not-found error, got {refused:?}");
	}

	/// A transaction belongs to the session that opened it. The map is
	/// process-wide, so without this check the id alone would let any caller
	/// read, write, commit or cancel inside somebody else's transaction.
	#[tokio::test]
	async fn a_transaction_is_only_usable_by_the_session_that_opened_it() {
		let service = service().await;
		let owner = Uuid::new_v4();
		service.rpc().attach(owner).await.expect("attach");
		let DbResult::Other(Value::Uuid(txn)) =
			service.rpc().begin(None, owner).await.expect("begin")
		else {
			panic!("begin should answer with a transaction id");
		};
		let txn = txn.into_inner();

		assert!(service.rpc().transaction_belongs_to(&txn, owner));
		assert!(!service.rpc().transaction_belongs_to(&txn, Uuid::new_v4()));
		assert!(!service.rpc().transaction_belongs_to(&Uuid::new_v4(), owner));
	}

	/// A transaction outlives the request that opened it, so it cannot run on
	/// the ephemeral session a request without one is given: releasing that
	/// session cancels the transaction the caller was just handed.
	#[tokio::test]
	async fn a_transaction_needs_an_attached_session() {
		let service = service().await;
		let ephemeral = Uuid::new_v4();
		service
			.rpc()
			.register_ephemeral_session(ephemeral, Arc::new(RwLock::new(Session::default())));

		let refused = service
			.rpc()
			.begin(None, ephemeral)
			.await
			.expect_err("an ephemeral session must not open a transaction");
		assert!(refused.is_validation(), "expected a validation error, got {refused:?}");
	}

	/// A subscriber that stops reading must be ended once its buffer fills.
	///
	/// Waiting for it instead would bound nothing: the notification dispatcher
	/// keeps receiving while a send is pending, so pending sends would
	/// accumulate without limit, each holding a whole notification.
	#[tokio::test]
	async fn a_subscriber_that_stops_reading_is_ended() {
		let service = service().await;
		let owner = Uuid::new_v4();
		let live_query_id = register_live_query(&service, owner).await;
		let subscription =
			service.attach_subscription(owner, live_query_id, false).expect("subscribe");
		// Held but never polled, so nothing drains the queue.
		let _stream = subscription.into_inner();

		let notification = Notification::new(
			live_query_id.into(),
			None,
			surrealdb_types::Action::Create,
			Value::None,
			Value::None,
		);
		// The buffer's worth of notifications fit; the one after it does not.
		for _ in 0..*GRPC_NOTIFICATION_BUFFER {
			assert!(service.rpc().dispatch_notification(&notification).await);
			assert!(
				service
					.rpc()
					.live_queries
					.get(&live_query_id)
					.is_some_and(|lq| lq.subscriber.is_some()),
				"the subscription must survive while its buffer has room"
			);
		}

		assert!(service.rpc().dispatch_notification(&notification).await);
		assert!(
			service
				.rpc()
				.live_queries
				.get(&live_query_id)
				.is_some_and(|lq| lq.subscriber.is_none()),
			"a full buffer must end the subscription rather than queue behind it"
		);
	}

	/// A gRPC session outlives any one connection, so nothing reclaims the
	/// transactions an abandoned one left open until the server stops.
	#[tokio::test]
	async fn shutdown_cancels_the_transactions_clients_left_open() {
		let service = service().await;
		let session_id = Uuid::new_v4();
		service.rpc().attach(session_id).await.expect("attach");
		service.rpc().begin(None, session_id).await.expect("begin");
		assert_eq!(service.rpc().transactions.len(), 1);

		service.rpc().cleanup_all_txns().await;
		assert!(service.rpc().transactions.is_empty(), "shutdown must cancel every transaction");
	}

	/// A session's teardown has to reap every live query that session holds and
	/// none that it does not: reaping fewer leaves a subscription delivering
	/// change data past the authorization that created it, reaping more crosses
	/// into another session's subscriptions.
	///
	/// It also has to release the session's count, or the counts would grow by
	/// one entry per session that ever registered a live query.
	#[tokio::test]
	async fn a_sessions_live_queries_are_reclaimed_and_no_others() {
		let service = service().await;
		let session_id = Uuid::new_v4();
		let first = register_live_query(&service, session_id).await;
		let second = register_live_query(&service, session_id).await;
		// Another session's, which this session's teardown must not touch.
		let bystander_session = Uuid::new_v4();
		let bystander = register_live_query(&service, bystander_session).await;

		let mut registered = vec![first, second];
		registered.sort();
		assert_eq!(
			registered_live_queries(&service, session_id),
			registered,
			"both live queries are registered under the session that made them"
		);
		assert_eq!(
			live_query_count(&service, session_id),
			Some(2),
			"and both are counted, so the session's teardown searches for them"
		);

		service.rpc().cleanup_lqs(&session_id).await;

		assert!(!service.rpc().live_queries.contains_key(&first), "the first is unregistered");
		assert!(!service.rpc().live_queries.contains_key(&second), "and so is the second");
		assert_eq!(
			live_query_count(&service, session_id),
			None,
			"the count entry is released rather than left behind at zero"
		);
		assert!(
			service.rpc().live_queries.contains_key(&bystander),
			"another session's live query survives"
		);
		assert_eq!(
			live_query_count(&service, bystander_session),
			Some(1),
			"and stays counted under its own session"
		);
	}

	/// The skip in a session's live-query teardown is only safe if a session
	/// holding a registration always holds a count: an uncounted registration
	/// would be skipped past and stranded, outliving the session it belongs
	/// to. Asserted in both directions, since a count that is never released
	/// costs every later teardown of that session a scan of the whole
	/// registry.
	#[tokio::test]
	async fn a_registered_live_query_is_counted_and_an_unregistered_one_is_not() {
		let service = service().await;
		let session_id = Uuid::new_v4();
		assert_eq!(
			live_query_count(&service, session_id),
			None,
			"a session that has registered nothing is not counted"
		);

		let lqid = register_live_query(&service, session_id).await;
		assert_eq!(
			live_query_count(&service, session_id),
			Some(1),
			"registering counts against the session"
		);

		service.rpc().forget_live_query(&lqid);
		assert_eq!(
			live_query_count(&service, session_id),
			None,
			"and unregistering releases the count"
		);
	}

	/// A registration whose session is already gone is discarded rather than
	/// published. This is the end state the register/detach race can produce,
	/// pinned deterministically: a live query left registered under a session
	/// that no longer exists would keep delivering change data under an
	/// authorization that has been torn down, and its datastore row would
	/// outlive every path that could find it.
	#[tokio::test]
	async fn a_live_query_whose_session_went_away_is_not_left_registered() {
		let service = service().await;
		// Never attached, which is the state a detach that has already run
		// leaves behind.
		let session_id = Uuid::new_v4();
		let lqid = Uuid::new_v4();

		service.rpc().handle_live(&lqid, session_id, None, None).await;

		assert!(
			!service.rpc().live_queries.contains_key(&lqid),
			"the registration must not survive the session it belongs to"
		);
		assert_eq!(live_query_count(&service, session_id), None, "and must leave no count behind");
	}

	/// Samples registration against removal, asserting what makes the count a
	/// safe skip-hint: it must never be absent while the registry still holds
	/// a registration for that session, because a teardown reading an absent
	/// count skips the scan that would have reaped it.
	///
	/// A sample, not a guard. Registration counts before it publishes, with no
	/// await point between, so the interleaving that would break the invariant
	/// is a few instructions wide and this passes whether or not the ordering
	/// is the right way round; it is here to catch a coarser regression, such
	/// as the counting being dropped from one of the two paths. The ordering
	/// each path relies on is documented on `live_query_counts`, and the
	/// count's two directions are pinned deterministically by
	/// [`a_registered_live_query_is_counted_and_an_unregistered_one_is_not`].
	#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
	async fn concurrent_registration_and_removal_keep_the_count_covering_the_registry() {
		let service = service().await;
		for _ in 0..2000 {
			let session_id = Uuid::new_v4();
			service.rpc().attach(session_id).await.expect("attach");
			let lqid = Uuid::new_v4();
			let g1 = Arc::clone(&service.state.grpc);
			let g2 = Arc::clone(&service.state.grpc);
			let register = tokio::spawn(async move {
				g1.handle_live(&lqid, session_id, None, None).await;
			});
			// Everything that ends a live query goes through here: a `KILL`, a
			// dropped table or database, a revoked principal.
			let end = tokio::spawn(async move {
				g2.forget_live_query(&lqid);
			});
			let _ = tokio::join!(register, end);

			// Whichever order they landed in, anything still registered under
			// the session must be covered by a count, or its teardown will
			// skip straight past it.
			let registered = registered_live_queries(&service, session_id);
			assert!(
				registered.is_empty() || live_query_count(&service, session_id).is_some(),
				"the registry holds {registered:?} for a session with no count"
			);
			// Leave the transport clean for the next iteration.
			service.rpc().cleanup_lqs(&session_id).await;
		}
	}

	/// `del_session` removes the session before sweeping what it owns, and
	/// registration counts before it publishes, so either can interleave with
	/// the other. Whichever way they land the session ends detached, so
	/// nothing may remain registered under it.
	#[tokio::test]
	async fn registering_a_live_query_racing_detach_strands_nothing() {
		let service = service().await;
		for _ in 0..100 {
			let session_id = Uuid::new_v4();
			service.rpc().attach(session_id).await.expect("attach a fresh session");
			let lqid = Uuid::new_v4();
			let g1 = Arc::clone(&service.state.grpc);
			let g2 = Arc::clone(&service.state.grpc);
			let register = tokio::spawn(async move {
				g1.handle_live(&lqid, session_id, None, None).await;
			});
			let detach = tokio::spawn(async move {
				let _ = g2.del_session(&session_id).await;
			});
			let _ = tokio::join!(register, detach);

			assert!(
				!service.rpc().live_queries.contains_key(&lqid),
				"live query {lqid} was stranded under detached session"
			);
			assert_eq!(
				live_query_count(&service, session_id),
				None,
				"a count was stranded under detached session"
			);
		}
	}

	/// A session's transaction teardown carries the same obligation as its
	/// live-query teardown: cancel every transaction that session opened and
	/// none that it did not. One missed would hold its locks past the session
	/// that opened it, until the server stops.
	#[tokio::test]
	async fn a_sessions_transactions_are_reclaimed_and_no_others() {
		let service = service().await;
		let session_id = Uuid::new_v4();
		service.rpc().attach(session_id).await.expect("attach");
		let bystander_session = Uuid::new_v4();
		service.rpc().attach(bystander_session).await.expect("attach");
		let txn = begin_txn(&service, session_id).await;
		let bystander = begin_txn(&service, bystander_session).await;

		assert!(
			service.rpc().transaction_counts.contains_key(&session_id),
			"an open transaction holds a slot, so the session's teardown searches for it"
		);

		service.rpc().cleanup_txns(&session_id).await;

		assert!(!service.rpc().transactions.contains_key(&txn), "the transaction is cancelled");
		assert!(
			!service.rpc().transaction_counts.contains_key(&session_id),
			"and its slot released"
		);
		assert!(
			service.rpc().transactions.contains_key(&bystander),
			"another session's transaction survives"
		);
		assert!(
			service.rpc().transaction_counts.contains_key(&bystander_session),
			"and keeps holding its own slot"
		);
	}

	/// `commit` and `cancel` end a transaction without going through any
	/// teardown, so they are the paths that have to release the session's slot
	/// themselves. One left reserved would spend the session's transaction
	/// budget on a transaction that has already finished.
	#[tokio::test]
	async fn finishing_a_transaction_releases_the_sessions_slot() {
		for (name, finish) in [("commit", true), ("cancel", false)] {
			let service = service().await;
			let session_id = Uuid::new_v4();
			service.rpc().attach(session_id).await.expect("attach");
			let txn = begin_txn(&service, session_id).await;

			let outcome = match finish {
				true => service.rpc().commit(Some(txn), session_id, Array::new()).await,
				false => service.rpc().cancel(Some(txn), session_id, Array::new()).await,
			};
			outcome.unwrap_or_else(|err| panic!("{name} should succeed: {err:?}"));

			assert!(
				!service.rpc().transaction_counts.contains_key(&session_id),
				"{name} must leave no slot reserved"
			);
			assert!(
				!service.rpc().transactions.contains_key(&txn),
				"{name} must remove the transaction"
			);
		}
	}

	/// `reset` sweeps what a session holds and leaves the session itself in
	/// place, so it is the one teardown that runs while the session can still
	/// be opening a transaction or registering a live query on another
	/// request. It must reclaim both.
	#[tokio::test]
	async fn reset_reclaims_what_the_session_holds_and_keeps_the_session() {
		let service = service().await;
		let session_id = Uuid::new_v4();
		service.rpc().attach(session_id).await.expect("attach");
		let txn = begin_txn(&service, session_id).await;
		let lqid = register_live_query(&service, session_id).await;

		service.rpc().reset(session_id).await.expect("reset");

		assert!(
			!service.rpc().transactions.contains_key(&txn),
			"reset must cancel the session's open transaction"
		);
		assert!(
			!service.rpc().live_queries.contains_key(&lqid),
			"reset must end the session's live query"
		);
		assert!(
			service.rpc().sessions.contains_key(&session_id),
			"and must leave the session itself attached"
		);
	}

	/// A live query is only reachable by the session that registered it.
	/// Without this, holding a live query id -- which is an ordinary query
	/// result, not a secret -- would be enough to receive another session's
	/// changes.
	///
	/// The refusal is `NotFound` rather than `PermissionDenied` so it does not
	/// confirm to a stranger that the id exists.
	#[tokio::test]
	async fn a_live_query_is_only_reachable_by_its_own_session() {
		let service = service().await;
		let owner = Uuid::new_v4();
		let live_query_id = register_live_query(&service, owner).await;

		let refused = service
			.attach_subscription(Uuid::new_v4(), live_query_id, false)
			.err()
			.expect("another session must be refused");
		assert_eq!(refused.code(), tonic::Code::NotFound);

		let Ok(_subscription) = service.attach_subscription(owner, live_query_id, false) else {
			panic!("the owner may subscribe");
		};
	}

	/// One subscriber at a time, matching what `GetCapabilities` reports. A
	/// second attach must be refused rather than quietly displacing the first,
	/// which would leave that subscriber with a stream that never ends and
	/// never delivers.
	#[tokio::test]
	async fn a_second_subscriber_is_refused() {
		let service = service().await;
		let owner = Uuid::new_v4();
		let live_query_id = register_live_query(&service, owner).await;

		let Ok(_subscription) = service.attach_subscription(owner, live_query_id, false) else {
			panic!("the first subscriber should be accepted");
		};
		let refused = service
			.attach_subscription(owner, live_query_id, false)
			.err()
			.expect("the second subscriber must be refused");
		assert_eq!(refused.code(), tonic::Code::AlreadyExists);
	}

	/// Dropping the stream is how a client unsubscribes, so it has to release
	/// the slot -- otherwise a client that reconnected could never re-attach
	/// to its own live query.
	#[tokio::test]
	async fn dropping_a_subscription_releases_it() {
		let service = service().await;
		let owner = Uuid::new_v4();
		let live_query_id = register_live_query(&service, owner).await;

		let Ok(subscription) = service.attach_subscription(owner, live_query_id, false) else {
			panic!("the first subscriber should be accepted");
		};
		drop(subscription);
		let Ok(_resubscribed) = service.attach_subscription(owner, live_query_id, false) else {
			panic!("re-subscribing after the first stream was dropped");
		};
	}

	/// Subscribing to a live query that does not exist is refused, so a client
	/// cannot register a subscription against an id nothing will ever produce.
	#[tokio::test]
	async fn an_unknown_live_query_is_refused() {
		let service = service().await;
		let refused = service
			.attach_subscription(Uuid::new_v4(), Uuid::new_v4(), false)
			.err()
			.expect("an unknown live query must be refused");
		assert_eq!(refused.code(), tonic::Code::NotFound);
	}

	/// An attached session a `LIVE SELECT` can run on: realtime, and pointed at
	/// a namespace and database. An ephemeral session is not marked realtime,
	/// so a streaming `LIVE SELECT` has to be given one of these.
	async fn realtime_session(service: &GrpcService) -> Uuid {
		let id = Uuid::new_v4();
		service.rpc().attach(id).await.expect("attach");
		let lock = service.rpc().get_session(&id).await.expect("the attached session");
		let mut session = lock.write().await;
		*session = Session::owner().with_ns("test").with_db("test").with_rt(true);
		session.id = Some(id);
		drop(session);
		id
	}

	fn context(session_id: Uuid) -> rpc::RequestContext {
		rpc::RequestContext {
			session: Some(proto::Uuid::from_uuid(session_id)),
			transaction: None,
			timeout: None,
		}
	}

	/// Writes to the table a test's live queries watch, from outside the
	/// transport so it works whether or not the request's session still exists.
	async fn write(service: &GrpcService, statement: &str) {
		let session = Session::owner().with_ns("test").with_db("test");
		let results =
			service.kvs().execute(statement, &session, None).await.expect("the write runs");
		for result in results {
			result.result.expect("the write succeeds");
		}
	}

	/// Creates the table these tests' live queries watch, and the namespace and
	/// database their sessions run in.
	async fn define_watched_table(service: &GrpcService) {
		write(service, "DEFINE NAMESPACE test; DEFINE DATABASE test; DEFINE TABLE thing").await;
	}

	/// Registers a live query in the datastore and nowhere else, which is the
	/// state a streaming query leaves one in until it finishes.
	async fn commit_live_query(service: &GrpcService) -> Uuid {
		let session = Session::owner().with_ns("test").with_db("test").with_rt(true);
		let mut results = service
			.kvs()
			.execute("LIVE SELECT * FROM thing", &session, None)
			.await
			.expect("the live query runs");
		match results.remove(0).result.expect("the live query succeeds") {
			Value::Uuid(id) => id.into_inner(),
			value => panic!("a live query answers with its id, got {value:?}"),
		}
	}

	/// How many live queries the datastore still holds against the watched
	/// table.
	///
	/// Read from the catalog rather than inferred from whether a change is
	/// notified: a writer's subscription list is cached against the table's
	/// committed live-query timestamp, and a deletion does not bump it, so
	/// notifications can outlive the row a caller is asking about.
	async fn live_query_rows(service: &GrpcService) -> usize {
		let session = Session::owner().with_ns("test").with_db("test");
		let mut results = service
			.kvs()
			.execute("INFO FOR TABLE thing", &session, None)
			.await
			.expect("the table info runs");
		let info = match results.remove(0).result.expect("the table info succeeds") {
			Value::Object(info) => info,
			value => panic!("table info answers with an object, got {value:?}"),
		};
		match info.get("lives") {
			Some(Value::Object(lives)) => lives.len(),
			other => panic!("table info reports its live queries, got {other:?}"),
		}
	}

	/// Reads frames until the one that completes a `LIVE SELECT`, which is
	/// where its row is committed and its id is on the wire -- and where a
	/// client that stops reading leaves the registration undone.
	async fn read_to_live_statement(frames: &mut ResponseStream<rpc::QueryResponse>) {
		loop {
			let frame = frames.next().await.expect("a frame").expect("a frame");
			if let Some(rpc::query_response::Frame::Batch(batch)) = &frame.frame
				&& batch.statement_kind == rpc::QueryStatementKind::Live as i32
			{
				return;
			}
		}
	}

	/// A client that stops reading abandons the query where it stands. A
	/// `LIVE SELECT` that had already committed has written a datastore row,
	/// and the registration that would make it reachable never runs: nothing
	/// can subscribe to it, no cleanup path can find it, and the broker keeps
	/// producing notifications for it that are discarded. The stream's own
	/// teardown is the last thing that knows the id, so it deletes the row.
	#[tokio::test]
	async fn an_abandoned_stream_deletes_the_live_query_it_committed() {
		let service = service().await;
		let session_id = realtime_session(&service).await;
		let context = context(session_id);
		define_watched_table(&service).await;

		// The `SLEEP` keeps the execution in flight after the `LIVE SELECT` has
		// committed, which is the window this is about: the statement's row is
		// written, and the registration that only happens once the whole query
		// finishes has not run.
		let response = service
			.stream_query(
				Some(&context),
				"LIVE SELECT * FROM thing; SLEEP 30s".to_string(),
				None,
				QUERY_BATCH_RECORDS,
			)
			.await
			.expect("the stream starts");
		let mut frames = response.into_inner();
		read_to_live_statement(&mut frames).await;
		assert_eq!(
			live_query_rows(&service).await,
			1,
			"the live query is committed before the stream is abandoned",
		);

		drop(frames);

		// The teardown hands the deletion to a task, so give it room to land.
		let mut deleted = false;
		for _ in 0..100 {
			if live_query_rows(&service).await == 0 {
				deleted = true;
				break;
			}
			tokio::time::sleep(Duration::from_millis(20)).await;
		}
		assert!(deleted, "an abandoned stream's live query must not outlive it");
		assert!(
			service.rpc().live_queries.is_empty(),
			"an abandoned stream registers nothing either",
		);
	}

	/// A run that fails as a whole carries no results, so there is nothing to
	/// register the live queries it had already committed from. They are
	/// deleted instead: a row nothing owns is the same leak an abandoned stream
	/// would have left.
	///
	/// The failure is injected rather than provoked: a statement's own error --
	/// a timeout, a rollback -- travels in that statement's frame and still
	/// completes the run, so what is exercised here is the narrower case the
	/// executor reports as no results at all.
	#[tokio::test]
	async fn a_failed_run_deletes_the_live_queries_it_committed() {
		let service = service().await;
		let session_id = realtime_session(&service).await;
		define_watched_table(&service).await;
		// A committed live query that no transport has registered, which is
		// what a stream holds once the statement that made it has finished.
		let id = commit_live_query(&service).await;
		assert_eq!(
			live_query_rows(&service).await,
			1,
			"the live query is committed before the run fails",
		);

		let (items, buffered) = bounded(QUERY_STREAM_BUFFER);
		for item in surrealdb_rpc::items_for_result(
			0,
			QueryResult {
				time: Duration::ZERO,
				result: Ok(Value::Uuid(surrealdb_types::Uuid::from(id))),
				query_type: QueryType::Live,
			},
		) {
			items.send(item).await.expect("the items fit the stream's buffer");
		}
		drop(items);
		let job = QueryStreamJob {
			statement_count: 1,
			run: Box::pin(async {
				Err(TypesError::internal("the run failed as a whole".to_string()))
			}),
		};
		let mut framing = QueryFraming::new(
			Arc::clone(&service.state),
			ResolvedSession {
				id: session_id,
				client: Some(session_id),
			},
			AuthPrincipalSnapshot::capture(&Session::default()),
			job,
			buffered,
			QUERY_BATCH_RECORDS,
			CancelHandle::new(),
		);

		// The failure ends the stream, and the deletion is awaited before that
		// frame is queued -- so by the time a client sees it, the row is gone.
		let mut failed = false;
		while let Some(frame) = framing.next().await {
			failed |= matches!(frame.frame, Some(rpc::query_response::Frame::Error(_)));
		}
		assert!(failed, "the run's failure ends the stream");
		assert_eq!(
			live_query_rows(&service).await,
			0,
			"a failed run's live query must not outlive it",
		);
		assert!(service.rpc().live_queries.is_empty(), "a failed run registers nothing either");
	}

	/// A session detached while its query ran cannot take a registration: the
	/// live query would deliver change data for an authorization that has been
	/// torn down, and the per-session cleanup that would have removed it has
	/// already run. The committed row is deleted instead.
	#[tokio::test]
	async fn a_detached_sessions_live_query_is_deleted_not_registered() {
		let service = service().await;
		let session_id = realtime_session(&service).await;
		let context = context(session_id);
		define_watched_table(&service).await;

		let response = service
			.stream_query(
				Some(&context),
				"LIVE SELECT * FROM thing; SLEEP 200ms".to_string(),
				None,
				QUERY_BATCH_RECORDS,
			)
			.await
			.expect("the stream starts");
		let mut frames = response.into_inner();
		read_to_live_statement(&mut frames).await;
		// Detach while the `SLEEP` holds the query open, so the session is gone
		// by the time the registration is attempted.
		service.rpc().detach(session_id).await.expect("detach the session");
		while let Some(frame) = frames.next().await {
			frame.expect("a frame");
		}

		assert!(
			service.rpc().live_queries.is_empty(),
			"a detached session's live query must not be registered",
		);
		assert_eq!(
			live_query_rows(&service).await,
			0,
			"a detached session's live query must not be left in the datastore",
		);
	}

	/// The framing layer is what sees a live query id go past: it rides the
	/// stream as an ordinary single value, and only the statement's terminal
	/// item says the statement that produced it was a `LIVE SELECT`.
	#[test]
	fn framing_reports_the_live_queries_it_frames() {
		let id = surrealdb_types::Uuid::new_v4();
		let mut framing = QueryFrames::new(QUERY_BATCH_RECORDS);
		let live = QueryResult {
			time: Duration::ZERO,
			result: Ok(Value::Uuid(id)),
			query_type: QueryType::Live,
		};
		let reported: Vec<Uuid> = surrealdb_rpc::items_for_result(0, live)
			.into_iter()
			.filter_map(|item| framing.absorb(item))
			.collect();
		assert_eq!(reported, vec![id.into_inner()], "the live query is reported once, on finish");

		// A statement that failed committed nothing, and an ordinary statement
		// that happens to return a uuid is not a live query.
		let mut framing = QueryFrames::new(QUERY_BATCH_RECORDS);
		let failed = QueryResult {
			time: Duration::ZERO,
			result: Err(TypesError::internal("no".to_string())),
			query_type: QueryType::Live,
		};
		let other = QueryResult {
			time: Duration::ZERO,
			result: Ok(Value::Uuid(surrealdb_types::Uuid::new_v4())),
			query_type: QueryType::Other,
		};
		let reported: Vec<Uuid> = [failed, other]
			.into_iter()
			.enumerate()
			.flat_map(|(index, result)| surrealdb_rpc::items_for_result(index, result))
			.filter_map(|item| framing.absorb(item))
			.collect();
		assert!(reported.is_empty(), "nothing else is taken for a live query: {reported:?}");
	}

	fn frames(results: Vec<QueryResult>) -> Vec<rpc::QueryResponse> {
		frames_with(results, QUERY_BATCH_RECORDS)
	}

	/// Frame finished results the way the streaming path frames them, so these
	/// tests exercise the framing the server actually serves.
	///
	/// Feeding whole results in is the same thing a statement that produced its
	/// rows all at once does, and `Begin` is prepended here because the handler
	/// sends it before the execution starts rather than as part of the framing.
	fn frames_with(results: Vec<QueryResult>, batch_records: usize) -> Vec<rpc::QueryResponse> {
		let mut framing = QueryFrames::new(batch_records);
		let statement_count = results.len() as u32;
		let mut frames = vec![rpc::QueryResponse {
			frame: Some(rpc::query_response::Frame::Begin(rpc::QueryBegin {
				query_id: Some(proto::Uuid::from_uuid(Uuid::new_v4())),
				statement_count,
			})),
		}];
		for (index, result) in results.into_iter().enumerate() {
			for item in surrealdb_rpc::items_for_result(index, result) {
				framing.absorb(item);
			}
		}
		framing.end(Duration::from_millis(3));
		while let Some(frame) = framing.pop() {
			frames.push(frame);
		}
		frames
	}

	fn batch(response: &rpc::QueryResponse) -> &rpc::QueryBatchFrame {
		match response.frame.as_ref() {
			Some(rpc::query_response::Frame::Batch(batch)) => batch,
			_ => panic!("expected a batch frame"),
		}
	}

	fn result(value: Value) -> QueryResult {
		QueryResult {
			time: Duration::from_millis(1),
			result: Ok(value),
			query_type: QueryType::Other,
		}
	}

	/// Nothing follows a statement's terminal batch, including rows the
	/// executor had already produced for it.
	///
	/// A value this server cannot encode terminates its statement, but the
	/// executor knows nothing of that and keeps sending the batches it is
	/// partway through. Framing those would put rows after the terminal batch,
	/// which the protocol forbids and a client acting on rows as they arrive
	/// would treat as real results.
	#[test]
	fn a_terminated_statement_takes_no_further_rows() {
		let mut frames = QueryFrames::new(QUERY_BATCH_RECORDS);
		frames.queue_terminal(
			0,
			rpc::QueryStatementKind::Other,
			Some(to_proto_error(&TypesError::internal("unencodable".to_string()))),
			0,
		);
		// The batches the executor was already partway through sending.
		for _ in 0..4 {
			frames.absorb(QueryStreamItem::Rows {
				index: 0,
				values: vec![Value::Bool(true); QUERY_BATCH_RECORDS],
			});
		}
		frames.absorb(QueryStreamItem::Finished {
			index: 0,
			time: Duration::ZERO,
			query_type: QueryType::Other,
			error: None,
		});
		frames.end(Duration::ZERO);

		let mut produced = Vec::new();
		while let Some(frame) = frames.pop() {
			produced.push(frame);
		}
		assert_eq!(produced.len(), 2, "one terminal batch and one end frame, nothing after");
		assert!(
			matches!(produced[0].frame, Some(rpc::query_response::Frame::Batch(_))),
			"the terminal batch"
		);
		assert!(matches!(produced[1].frame, Some(rpc::query_response::Frame::End(_))));
	}

	/// The deadline a streaming query runs under is the sooner of the client's
	/// own and the server's, and either alone still applies.
	///
	/// This is what bounds a client that stops reading: its execution is polled
	/// only when it reads, so the deadline has to come from a timer the runtime
	/// drives, and that timer needs a duration whichever side supplied one.
	#[test]
	fn a_streaming_query_takes_the_sooner_deadline() {
		let short = Duration::from_secs(1);
		let long = Duration::from_secs(60);
		assert_eq!(shortest(Some(short), Some(long)), Some(short));
		assert_eq!(shortest(Some(long), Some(short)), Some(short));
		assert_eq!(shortest(Some(short), None), Some(short), "the client's alone still applies");
		assert_eq!(shortest(None, Some(long)), Some(long), "the server's alone still applies");
		assert_eq!(shortest(None, None), None, "neither configured means no deadline");
	}

	/// A statement is terminated once, whichever ends it first.
	///
	/// A value this server cannot encode terminates its statement immediately,
	/// and the executor -- which knows nothing of the encoding -- goes on to
	/// send that statement's own `Finished`. Acting on both would put a frame
	/// after a terminal one and count the statement twice in `result_count`,
	/// which a client cross-checking that count reads as a lost frame.
	#[test]
	fn a_statement_is_terminated_only_once() {
		let mut frames = QueryFrames::new(QUERY_BATCH_RECORDS);
		frames.queue_terminal(0, rpc::QueryStatementKind::Other, None, 0);
		frames.absorb(QueryStreamItem::Finished {
			index: 0,
			time: Duration::ZERO,
			query_type: QueryType::Other,
			error: None,
		});
		frames.end(Duration::ZERO);

		let mut produced = Vec::new();
		while let Some(frame) = frames.pop() {
			produced.push(frame);
		}
		assert_eq!(produced.len(), 2, "one terminal batch and one end frame");
		assert!(matches!(produced[0].frame, Some(rpc::query_response::Frame::Batch(_))));
		match produced[1].frame.as_ref() {
			Some(rpc::query_response::Frame::End(end)) => {
				assert_eq!(end.result_count, 1, "the statement is counted once");
			}
			other => panic!("expected an end frame, got {other:?}"),
		}
	}

	/// A query stream always opens with `Begin` and closes with `End`. This
	/// server executes to completion first, so every statement it parsed also
	/// produced a result: `Begin.statement_count` and `End.result_count` agree,
	/// and `End` reports how long the whole query took.
	#[test]
	fn query_frames_are_bracketed_by_begin_and_end() {
		let frames = frames(vec![result(Value::None), result(Value::None)]);
		assert_eq!(frames.len(), 4);
		match frames[0].frame.as_ref() {
			Some(rpc::query_response::Frame::Begin(begin)) => {
				assert_eq!(begin.statement_count, 2);
			}
			_ => panic!("expected a begin frame"),
		}
		match frames[3].frame.as_ref() {
			Some(rpc::query_response::Frame::End(end)) => {
				assert_eq!(end.result_count, 2, "every parsed statement produced a result");
				assert!(
					end.execution_duration.is_some(),
					"the whole-query duration is what a client cannot sum from the per-statement ones"
				);
			}
			_ => panic!("expected an end frame"),
		}
	}

	/// A client's batch size is honoured, and a request above the server's own
	/// bound is clamped rather than refused -- asking for more than you may have
	/// still returns results.
	#[test]
	fn a_client_batch_size_is_honoured_and_clamped() {
		assert_eq!(batch_records(0), QUERY_BATCH_RECORDS, "zero means the server chooses");
		assert_eq!(batch_records(1), 1, "a client may ask for one record per batch");
		assert_eq!(batch_records(u32::MAX), QUERY_BATCH_RECORDS, "clamped, not refused");

		let rows = Array::from(
			(0..3).map(|i| Value::Number(surrealdb_types::Number::Int(i))).collect::<Vec<_>>(),
		);
		let frames = frames_with(vec![result(Value::Array(rows))], batch_records(1));
		let sizes = batch_sizes(&frames);
		assert!(
			sizes.iter().all(|n| *n <= 1),
			"a client asking for one record per batch gets no more than one, {sizes:?}"
		);
		assert_eq!(sizes.iter().sum::<usize>(), 3, "every record is still sent once");
	}

	/// How many records each of a statement's batches carried.
	fn batch_sizes(frames: &[rpc::QueryResponse]) -> Vec<usize> {
		frames[1..frames.len() - 1]
			.iter()
			.map(|response| match batch(response).payload.as_ref() {
				Some(rpc::query_batch_frame::Payload::Values(values)) => values.values.len(),
				None => 0,
				_ => panic!("expected a value batch"),
			})
			.collect()
	}

	/// The reserve exists to cover what the transport adds to a payload, and
	/// the larger term of that is compression, which is proportional. tonic
	/// weighs the compressed payload against its send limit and compresses
	/// unconditionally once an encoding is negotiated, so data that does not
	/// compress grows: measured on incompressible input, gzip at level 6 adds
	/// one byte per ~3200 and zstd one per ~43000, both of it stored-block
	/// framing.
	///
	/// The reserve therefore has to stay clear of that rate at every size a
	/// deployment can configure, which a fixed number cannot do. One byte per
	/// 2048 is worse than either codec and is the rate asserted here.
	#[test]
	fn the_transport_reserve_clears_worst_case_compression() {
		for limit in [
			crate::cnf::GRPC_MIN_MESSAGE_SIZE,
			4 << 20,
			128 << 20,
			crate::cnf::GRPC_MAX_MESSAGE_CEILING,
		] {
			let payload = limit - transport_reserve(limit);
			assert!(
				payload + payload / 2048 <= limit,
				"a {payload} byte payload under a {limit} byte limit has no room to expand"
			);
		}
	}

	/// The record count is only half the bound. Records wide enough that a full
	/// batch of them would not fit have to be split on bytes instead, or the
	/// frame is one neither peer will take.
	///
	/// The failure this guards is silent when responses are compressed: tonic
	/// weighs the compressed payload against its send limit, so an oversized
	/// frame that happens to compress goes out and dies in the client's
	/// decompressor, naming no size at all.
	#[test]
	fn a_frame_of_wide_records_is_split_on_bytes() {
		use surrealdb_types::prost::Message;

		// Narrow enough that a frame packs tight against the budget rather than
		// stopping a whole record short of it, and wide enough that the byte
		// bound is still what closes the frame rather than the record count. A
		// fixture that leaves a record's width of slack cannot tell a budget
		// holding back a fixed kibibyte from one holding back a proportional
		// share -- the two differ by less than one wide record.
		let width = *GRPC_MAX_MESSAGE_SIZE / QUERY_BATCH_RECORDS;
		// Enough to reach a full frame through the ramp: 16, 32, 64 and 128
		// records go out before one is bounded by bytes rather than by count.
		let records = 16 + 32 + 64 + 128 + QUERY_BATCH_RECORDS;
		let rows =
			Array::from((0..records).map(|_| Value::String("x".repeat(width))).collect::<Vec<_>>());
		let frames = frames(vec![result(Value::Array(rows))]);

		let batches: Vec<&rpc::QueryBatchFrame> =
			frames[1..frames.len() - 1].iter().map(batch).collect();
		let framed: usize = batches
			.iter()
			.map(|b| match &b.payload {
				Some(rpc::query_batch_frame::Payload::Values(values)) => values.values.len(),
				_ => 0,
			})
			.sum();
		assert_eq!(framed, records, "every record still arrives, just in more frames");
		assert!(
			batches.len() > records.div_ceil(QUERY_BATCH_RECORDS),
			"the record count alone would not have split this: {} frames",
			batches.len()
		);

		let largest = frames.iter().map(|frame| frame.encoded_len()).max().expect("frames");
		// Without this the rest proves nothing: a fixture whose frames all sit
		// comfortably below the budget passes whatever the budget is.
		assert!(
			largest + width >= frame_payload_budget(),
			"the fixture has to fill a frame for this to test the budget: largest is {largest} \
			 against a {} byte budget",
			frame_payload_budget()
		);
		for frame in &frames {
			// The bound that matters is the one after compression: tonic weighs
			// the compressed payload, and data that does not compress grows.
			let encoded = frame.encoded_len();
			assert!(
				encoded + encoded / 256 <= *GRPC_MAX_MESSAGE_SIZE,
				"a frame encodes to {encoded} bytes, which worst-case expansion takes past the \
				 {} byte limit both peers enforce",
				*GRPC_MAX_MESSAGE_SIZE
			);
		}
	}

	/// A record wider than a whole message has no batch size that would carry
	/// it, so its statement fails saying so. Emitting it regardless would put a
	/// frame on the wire that the client cannot decode, which costs the caller
	/// the rest of the query as well.
	#[test]
	fn a_record_no_frame_can_carry_fails_its_statement() {
		let rows = Array::from(vec![
			Value::String("x".to_string()),
			Value::String("x".repeat(*GRPC_MAX_MESSAGE_SIZE)),
		]);
		let frames = frames(vec![result(Value::Array(rows))]);
		let batches: Vec<&rpc::QueryBatchFrame> =
			frames[1..frames.len() - 1].iter().map(batch).collect();

		let terminal = batches.last().expect("a terminal batch");
		let error = terminal.error.as_ref().expect("the statement fails");
		assert_eq!(error.kind, proto::ErrorKind::Validation as i32);
		assert!(
			error.message.contains("above the"),
			"the refusal has to name the limit: {}",
			error.message
		);
		assert!(
			batches.iter().all(|b| b.payload.is_none()
				|| matches!(
					&b.payload,
					Some(rpc::query_batch_frame::Payload::Values(v)) if v.values.is_empty()
				)),
			"a failed statement retracts its rows rather than framing them beside the refusal"
		);
	}

	/// An error is the one frame whose size a query decides: `THROW` puts the
	/// thrown value in the message. Trimming it is what keeps the subscriber or
	/// caller an error at all -- an oversized one is not a large error, it is
	/// no error and a broken stream.
	#[test]
	fn an_oversized_error_is_trimmed_rather_than_lost() {
		use surrealdb_types::prost::Message;

		const BUDGET: usize = 512;
		let thrown = TypesError::thrown("y".repeat(BUDGET * 4))
			.with_cause(TypesError::internal("z".repeat(BUDGET * 4)));
		let bounded = bounded_error(proto_error(&thrown), BUDGET);

		assert!(
			bounded.encoded_len() <= BUDGET,
			"trimmed to {} bytes against a {BUDGET} byte budget",
			bounded.encoded_len()
		);
		assert_eq!(
			bounded.kind,
			proto::ErrorKind::Thrown as i32,
			"the kind is what a client branches on and survives every trim"
		);
		assert!(
			bounded.message.ends_with(TRUNCATION_MARKER),
			"a trimmed message says it was trimmed: {}",
			bounded.message
		);
		assert!(bounded.cause.is_none(), "the cause chain goes before the message");

		// An error that already fits is passed through untouched.
		let small = proto_error(&TypesError::thrown("nope".to_string()));
		assert_eq!(bounded_error(small.clone(), BUDGET), small);
	}

	/// A result larger than one batch is split, so no single message grows with
	/// the result set -- a client decoding with gRPC's 4 MiB default would
	/// otherwise refuse a large `SELECT` outright.
	///
	/// The batches ramp rather than being uniform: the first is small so the
	/// client sees rows early, and each one doubles up to the cap so a long
	/// result does not pay a per-frame cost for that.
	#[test]
	fn a_large_result_is_split_across_batches() {
		let records = QUERY_BATCH_RECORDS * 2 + 1;
		let rows = Array::from(
			(0..records)
				.map(|i| Value::Number(surrealdb_types::Number::Int(i as i64)))
				.collect::<Vec<_>>(),
		);
		let frames = frames(vec![result(Value::Array(rows))]);
		let batches: Vec<&rpc::QueryBatchFrame> =
			frames[1..frames.len() - 1].iter().map(batch).collect();

		assert!(batches.len() > 1, "a result this size must span several batches");
		assert_eq!(
			batches.iter().map(|b| b.batch_index).collect::<Vec<_>>(),
			(0..batches.len() as u64).collect::<Vec<_>>(),
			"batches must be indexed in order so the client can demultiplex them"
		);
		let kinds: Vec<i32> = batches.iter().map(|b| b.kind).collect();
		let (last, rest) = kinds.split_last().expect("at least one batch");
		assert_eq!(*last, rpc::QueryResponseKind::BatchedFinal as i32);
		assert!(
			rest.iter().all(|k| *k == rpc::QueryResponseKind::Batched as i32),
			"only the last batch completes the statement"
		);

		let sizes = batch_sizes(&frames);
		assert_eq!(sizes.iter().sum::<usize>(), records, "every record must be sent once");
		assert_eq!(sizes[0], QUERY_FIRST_BATCH_RECORDS, "the first batch is small, for latency");
		assert!(
			sizes.windows(2).all(|w| w[1] >= w[0] || w[1] == sizes[sizes.len() - 1]),
			"batches grow toward the cap, {sizes:?}"
		);
		assert!(sizes.iter().all(|n| *n <= QUERY_BATCH_RECORDS), "no batch exceeds the cap");

		// The stats describe the statement, so they ride on the batch that
		// completes it rather than being repeated or split.
		assert!(rest.iter().enumerate().all(|(i, _)| batches[i].stats.is_none()));
		assert_eq!(
			batches
				.last()
				.expect("at least one batch")
				.stats
				.as_ref()
				.expect("the final batch carries the stats")
				.records_returned,
			records as i64
		);
	}

	/// An empty list still owes the client one final batch, so it learns the
	/// statement's kind and stats.
	#[test]
	fn an_empty_result_still_sends_one_final_batch() {
		let frames = frames(vec![result(Value::Array(Array::new()))]);
		assert_eq!(frames.len(), 3);
		let batch = batch(&frames[1]);
		assert_eq!(batch.kind, rpc::QueryResponseKind::BatchedFinal as i32);
		assert_eq!(batch.stats.as_ref().expect("stats").records_returned, 0);
	}

	/// A list result is a batch of its elements; the SDK rebuilds the array
	/// from them. A scalar is a `SINGLE`, which the SDK unwraps.
	#[test]
	fn list_and_scalar_results_use_distinct_kinds() {
		let list = frames(vec![result(Value::Array(Array::from(vec![
			Value::Bool(true),
			Value::Bool(false),
		])))]);
		let list = batch(&list[1]);
		assert_eq!(list.kind, rpc::QueryResponseKind::BatchedFinal as i32);
		assert_eq!(list.stats.as_ref().expect("stats").records_returned, 2);

		let scalar = frames(vec![result(Value::Bool(true))]);
		let scalar = batch(&scalar[1]);
		assert_eq!(scalar.kind, rpc::QueryResponseKind::Single as i32);
		assert_eq!(scalar.stats.as_ref().expect("stats").records_returned, 1);
	}

	/// A statement's own failure rides in its batch rather than terminating
	/// the stream, so the statements around it still report their results.
	#[test]
	fn a_failed_statement_does_not_end_the_stream() {
		let frames = frames(vec![
			QueryResult {
				time: Duration::ZERO,
				result: Err(TypesError::query("boom".to_string(), None)),
				query_type: QueryType::Other,
			},
			result(Value::Bool(true)),
		]);
		let failed = batch(&frames[1]);
		assert_eq!(failed.error.as_ref().expect("error").kind, proto::ErrorKind::Query as i32);
		assert!(failed.payload.is_none());
		assert!(batch(&frames[2]).error.is_none());
		assert!(matches!(frames[3].frame, Some(rpc::query_response::Frame::End(_))));
	}

	/// The statement kind lets a client tell a LIVE SELECT's returned id from
	/// an ordinary result.
	#[test]
	fn live_and_kill_statements_are_labelled() {
		let live = frames(vec![QueryResult {
			time: Duration::ZERO,
			result: Ok(Value::None),
			query_type: QueryType::Live,
		}]);
		assert_eq!(batch(&live[1]).statement_kind, rpc::QueryStatementKind::Live as i32);

		let kill = frames(vec![QueryResult {
			time: Duration::ZERO,
			result: Ok(Value::None),
			query_type: QueryType::Kill,
		}]);
		assert_eq!(batch(&kill[1]).statement_kind, rpc::QueryStatementKind::Kill as i32);
	}

	/// `use` distinguishes "leave alone" from "clear", which a plain string
	/// cannot express.
	#[test]
	fn nullable_strings_are_three_state() {
		assert_eq!(from_nullable(None), Value::None);
		assert_eq!(
			from_nullable(Some(rpc::NullableString {
				value: Some(rpc::nullable_string::Value::Null(proto::NullValue {})),
			})),
			Value::Null
		);
		assert_eq!(
			from_nullable(Some(rpc::NullableString {
				value: Some(rpc::nullable_string::Value::Some("test".to_string())),
			})),
			Value::String("test".to_string())
		);
	}

	/// Credentials round-trip to the shape the `signin` method parses: the
	/// scope fields are named, and empty ones are omitted rather than sent as
	/// empty strings (which would select a namespace called "").
	#[test]
	fn user_credentials_flatten_to_the_signin_object() {
		let object = access_credentials(rpc::AccessMethod {
			method: Some(rpc::access_method::Method::User(rpc::UserCredentials {
				namespace: String::new(),
				database: String::new(),
				username: "root".to_string(),
				password: "root".to_string(),
				access: String::new(),
			})),
		})
		.expect("credentials");
		assert_eq!(object.get("user"), Some(&Value::String("root".to_string())));
		assert_eq!(object.get("pass"), Some(&Value::String("root".to_string())));
		assert!(object.get("ns").is_none());
		assert!(object.get("db").is_none());
	}

	/// Record credentials carry the access method's own variables through to
	/// the SIGNIN/SIGNUP clause alongside the scope fields.
	#[test]
	fn record_credentials_carry_their_variables() {
		let variables: proto::Object = std::collections::BTreeMap::from([(
			"email".to_string(),
			proto::Value::try_from(Value::String("a@b.c".to_string())).expect("encodable"),
		)])
		.into();
		let object = record_credentials(rpc::RecordCredentials {
			namespace: "test".to_string(),
			database: "test".to_string(),
			access: "user".to_string(),
			variables: Some(proto::Variables {
				variables: variables.items,
			}),
		})
		.expect("credentials");
		assert_eq!(object.get("ns"), Some(&Value::String("test".to_string())));
		assert_eq!(object.get("ac"), Some(&Value::String("user".to_string())));
		assert_eq!(object.get("email"), Some(&Value::String("a@b.c".to_string())));
	}

	/// An export config with no table selection exports every table, matching
	/// the datastore's own default.
	#[test]
	fn export_table_selection_maps_each_arm() {
		let config = |tables: Option<rpc::export_config::Tables>| {
			from_proto_export_config(rpc::ExportConfig {
				tables,
				..Default::default()
			})
			.tables
		};
		assert!(matches!(config(None), export::TableConfig::All));
		assert!(matches!(
			config(Some(rpc::export_config::Tables::from(false))),
			export::TableConfig::None
		));
		match config(Some(rpc::export_config::Tables {
			selection: Some(rpc::export_config::tables::Selection::Selected(
				rpc::export_config::SelectedTables {
					tables: vec!["person".to_string()],
				},
			)),
		})) {
			export::TableConfig::Some(tables) => assert_eq!(tables, vec!["person".to_string()]),
			other => panic!("expected a table selection, got {other:?}"),
		}
	}

	/// Every error kind maps to the status code the SDK's engine maps back
	/// onto that same kind, so a failure keeps its classification across the
	/// wire even though a status carries no structured error.
	#[test]
	fn error_kinds_map_to_round_tripping_status_codes() {
		use tonic::Code;
		let cases = [
			(TypesError::validation("v".to_string(), None), Code::InvalidArgument),
			(TypesError::configuration("c".to_string(), None), Code::Unimplemented),
			(TypesError::query("q".to_string(), None), Code::Aborted),
			(TypesError::not_allowed("n".to_string(), None), Code::PermissionDenied),
			(TypesError::not_found("f".to_string(), None), Code::NotFound),
			(TypesError::already_exists("a".to_string(), None), Code::AlreadyExists),
			(TypesError::connection("x".to_string(), None), Code::Unavailable),
			(TypesError::internal("i".to_string()), Code::Internal),
		];
		for (error, code) in cases {
			assert_eq!(to_status(&error).code(), code, "{}", error.message());
		}
	}

	/// The structured form keeps the cause chain that the status form has
	/// nowhere to put.
	#[test]
	fn proto_errors_keep_their_cause_chain() {
		let error = TypesError::query("outer".to_string(), None).with_cause(
			TypesError::serialization("inner".to_string(), SerializationError::Deserialization),
		);
		let proto = to_proto_error(&error);
		assert_eq!(proto.kind, proto::ErrorKind::Query as i32);
		let cause = proto.cause.expect("cause");
		assert_eq!(cause.kind, proto::ErrorKind::Serialization as i32);
		assert_eq!(cause.message, "inner");
	}

	/// The reason and its payload are what a client is left to classify on:
	/// the protocol states the message is not part of the contract and will be
	/// reworded, so an error carrying only a kind gives a client nothing stable
	/// to branch on.
	#[test]
	fn proto_errors_carry_their_reason() {
		// Details nest: the reason sits one level below the kind, and the
		// content one level below that.
		let nested = to_proto_error(&TypesError::not_allowed(
			"nope".to_string(),
			NotAllowedError::Auth(AuthError::TokenExpired),
		));
		assert_eq!(nested.kind, proto::ErrorKind::NotAllowed as i32);
		assert_eq!(nested.detail_kind(), Some("Auth"));
		let content = nested.details.and_then(|details| details.content).expect("nested content");
		assert_eq!(
			Value::try_from(content).expect("a decodable payload"),
			AuthError::TokenExpired.into_value()
		);

		// A leaf payload travels in the same field as a nested reason does.
		let timed_out = to_proto_error(&TypesError::query(
			"too slow".to_string(),
			QueryError::TimedOut {
				duration: Duration::from_secs(5),
			},
		));
		assert_eq!(timed_out.detail_kind(), Some("TimedOut"));
		assert!(timed_out.details.and_then(|details| details.content).is_some());

		// An error whose kind is its whole story carries no reason rather than
		// an empty one, which is what lets a client tell the two apart.
		assert_eq!(to_proto_error(&TypesError::internal("i".to_string())).detail_kind(), None);
	}

	/// A conflict is the one failure the engine reports as transient, and the
	/// hint's presence is the entire signal: absence has to read as "not
	/// retryable", never as "the server did not say".
	#[test]
	fn a_conflict_is_the_only_error_marked_retryable() {
		let conflict = to_proto_error(&TypesError::query(
			"This transaction can be retried".to_string(),
			QueryError::TransactionConflict,
		));
		assert!(conflict.is_retryable());
		assert_eq!(conflict.detail_kind(), Some("TransactionConflict"));

		for error in [
			TypesError::query("cancelled".to_string(), QueryError::Cancelled),
			TypesError::query("no reason given".to_string(), None),
			TypesError::internal("broken".to_string()),
		] {
			assert!(!to_proto_error(&error).is_retryable(), "{}", error.message());
		}
	}

	/// A client sizes its own decode and encode limits from what the handshake
	/// reports, so the advertised size has to be the one the route enforces.
	/// Advertising more invites a message the service then rejects mid-body,
	/// which reaches the client as an opaque stream error.
	#[tokio::test]
	async fn the_advertised_message_size_is_the_configured_one() {
		let limits = service().await.server_capabilities().limits.expect("limits");
		assert_eq!(limits.max_message_bytes, *GRPC_MAX_MESSAGE_SIZE as u64);
	}

	/// The floor on that size exists so a mistaken configuration degrades to a
	/// working service rather than one that starts and then fails every call.
	/// It only does that if the handshake itself fits inside it -- a client
	/// that cannot read the capabilities never gets as far as any other call.
	#[tokio::test]
	async fn the_smallest_allowed_message_size_still_carries_the_handshake() {
		use surrealdb_types::prost::Message;

		let capabilities = service().await.server_capabilities();
		assert!(
			capabilities.encoded_len() < crate::cnf::GRPC_MIN_MESSAGE_SIZE,
			"the handshake encodes to {} bytes, which the {} byte floor cannot carry",
			capabilities.encoded_len(),
			crate::cnf::GRPC_MIN_MESSAGE_SIZE
		);
	}

	/// The tokens an auth method answered with are read from whichever shape
	/// it used: a bare token, or an object carrying a refresh token too.
	#[test]
	fn tokens_are_read_from_either_answer_shape() {
		let bare = to_tokens(DbResult::Other(Value::String("access".to_string()))).expect("tokens");
		assert_eq!(bare.access, "access");
		assert!(bare.refresh.is_empty());

		let mut object = Object::new();
		object.insert("access".to_string(), Value::String("access".to_string()));
		object.insert("refresh".to_string(), Value::String("refresh".to_string()));
		let pair = to_tokens(DbResult::Other(Value::Object(object))).expect("tokens");
		assert_eq!(pair.access, "access");
		assert_eq!(pair.refresh, "refresh");
	}

	/// A trailer is what marks a byte stream complete, so a mismatched length
	/// or checksum has to fail the transfer rather than be ignored.
	#[test]
	fn trailers_are_verified_against_what_arrived() {
		let digest = blake3::hash(b"hello");
		let trailer = rpc::DataTrailer {
			bytes: 5,
			blake3: digest.to_hex().to_string(),
		};
		assert!(verify_trailer(&trailer, 5, digest).is_ok());
		assert!(verify_trailer(&trailer, 4, digest).is_err());

		// An absent checksum means the sender did not compute one.
		let unchecked = rpc::DataTrailer {
			bytes: 5,
			blake3: String::new(),
		};
		assert!(verify_trailer(&unchecked, 5, digest).is_ok());

		let wrong = rpc::DataTrailer {
			bytes: 5,
			blake3: blake3::hash(b"other").to_hex().to_string(),
		};
		assert!(verify_trailer(&wrong, 5, digest).is_err());
	}
}