rivetkit_core/actor/sqlite/

tx.rs

use std::{
	collections::{BTreeMap, VecDeque},
	error::Error,
	fmt,
	future::Future,
	sync::{
		Arc,
		atomic::{AtomicU64, Ordering},
	},
	time::Duration,
};

use anyhow::{Context, Result};
use serde::Serialize;
#[cfg(target_arch = "wasm32")]
use tokio::sync::oneshot;
use tokio::sync::{
	Mutex as AsyncMutex, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, OwnedSemaphorePermit, RwLock,
	Semaphore, TryAcquireError,
};
use tokio_util::sync::CancellationToken;

#[cfg(not(target_arch = "wasm32"))]
use crate::runtime::RuntimeSpawner;

use super::{BindParam, ExecuteResult, QueryResult, SqliteDb, report_sqlite_worker_fatal};

pub const DEFAULT_TRANSACTION_TIMEOUT: Duration = Duration::from_secs(60);
pub const TRANSACTION_COORDINATOR_QUEUE_CAPACITY: usize = 128;
pub(super) const TRANSACTION_TERMINAL_CAPACITY: usize = 1024;

#[derive(Clone)]
pub struct SqliteTransaction {
	db: SqliteDb,
	key: String,
}

impl SqliteTransaction {
	pub fn key(&self) -> &str {
		&self.key
	}

	pub async fn exec(&self, sql: impl Into<String>) -> Result<QueryResult> {
		self.db.transaction_exec(&self.key, sql.into()).await
	}

	pub async fn execute(
		&self,
		sql: impl Into<String>,
		params: Option<Vec<BindParam>>,
	) -> Result<ExecuteResult> {
		self.db
			.transaction_execute(&self.key, sql.into(), params)
			.await
	}

	pub async fn commit(&self) -> Result<()> {
		self.db.finish_transaction(&self.key, true).await
	}

	pub async fn rollback(&self) -> Result<()> {
		self.db.finish_transaction(&self.key, false).await
	}

	pub async fn expire(&self) -> Result<()> {
		self.db.expire_transaction(&self.key).await
	}
}

pub(super) struct TransactionCoordinator {
	pub(super) gate: Arc<RwLock<()>>,
	pub(super) admission: Arc<Semaphore>,
	pub(super) state: AsyncMutex<TransactionCoordinatorState>,
	epoch: AtomicU64,
}

pub(super) struct TransactionCoordinatorState {
	pub(super) active: Option<ActiveTransaction>,
	pub(super) terminal: BTreeMap<String, TransactionTerminalState>,
	pub(super) terminal_order: VecDeque<String>,
	pub(super) poisoned: BTreeMap<String, Duration>,
	pub(super) last_expired_timeout: Option<Duration>,
	pub(super) closed: bool,
}

pub(super) struct ActiveTransaction {
	key: String,
	timeout: Duration,
	pub(super) expiring: bool,
	connection_lost: bool,
	remote_session: Option<u64>,
	gate_guard: Option<OwnedRwLockWriteGuard<()>>,
	operation: Arc<AsyncMutex<()>>,
	timeout_task: Option<CancellationToken>,
}

#[derive(Clone, Copy)]
pub(super) enum TransactionTerminalState {
	Committed,
	RolledBack,
	Expired(Duration),
	ConnectionLost,
}

pub(super) struct RegularOperationGuard {
	_gate: OwnedRwLockReadGuard<()>,
	_permit: OwnedSemaphorePermit,
}

impl Default for TransactionCoordinator {
	fn default() -> Self {
		Self {
			gate: Arc::new(RwLock::new(())),
			admission: Arc::new(Semaphore::new(TRANSACTION_COORDINATOR_QUEUE_CAPACITY)),
			state: AsyncMutex::new(TransactionCoordinatorState {
				active: None,
				terminal: BTreeMap::new(),
				terminal_order: VecDeque::new(),
				poisoned: BTreeMap::new(),
				last_expired_timeout: None,
				closed: false,
			}),
			epoch: AtomicU64::new(0),
		}
	}
}

impl SqliteDb {
	pub(super) fn try_transaction_admission(&self) -> Result<OwnedSemaphorePermit> {
		match Arc::clone(&self.transaction_coordinator.admission).try_acquire_owned() {
			Ok(permit) => Ok(permit),
			Err(TryAcquireError::NoPermits) => Err(transaction_queue_full_error()),
			Err(TryAcquireError::Closed) => Err(transaction_coordinator_closed_error()),
		}
	}

	pub(super) async fn begin_regular_operation(&self) -> Result<RegularOperationGuard> {
		let epoch = self.transaction_coordinator.epoch.load(Ordering::Acquire);
		let permit = self.try_transaction_admission()?;
		let gate = Arc::clone(&self.transaction_coordinator.gate)
			.read_owned()
			.await;
		let state = self.transaction_coordinator.state.lock().await;
		if state.closed {
			return Err(transaction_coordinator_closed_error());
		}
		if self.transaction_coordinator.epoch.load(Ordering::Acquire) != epoch {
			return Err(transaction_expired_error(
				state
					.last_expired_timeout
					.unwrap_or(DEFAULT_TRANSACTION_TIMEOUT),
			));
		}
		drop(state);
		Ok(RegularOperationGuard {
			_gate: gate,
			_permit: permit,
		})
	}

	pub async fn begin_transaction(&self, timeout: Option<Duration>) -> Result<SqliteTransaction> {
		self.begin_transaction_with_key(uuid::Uuid::new_v4().to_string(), timeout)
			.await
	}

	pub async fn begin_transaction_with_key(
		&self,
		key: impl Into<String>,
		timeout: Option<Duration>,
	) -> Result<SqliteTransaction> {
		let key = key.into();
		let timeout = timeout.unwrap_or(DEFAULT_TRANSACTION_TIMEOUT);
		if key.is_empty() {
			return Err(transaction_invalid_argument_error(
				"transaction key must not be empty",
			));
		}
		if timeout.is_zero() {
			return Err(transaction_invalid_argument_error(
				"transaction timeout must be greater than zero",
			));
		}

		let db = self.clone();
		run_detached_transaction_task(
			async move { db.begin_transaction_inner(key, timeout).await },
			"sqlite transaction begin task failed",
		)
		.await
	}

	pub(super) async fn begin_transaction_inner(
		&self,
		key: String,
		timeout: Duration,
	) -> Result<SqliteTransaction> {
		let epoch = self.transaction_coordinator.epoch.load(Ordering::Acquire);
		let permit = self.try_transaction_admission()?;
		let gate_guard = Arc::clone(&self.transaction_coordinator.gate)
			.write_owned()
			.await;
		{
			let state = self.transaction_coordinator.state.lock().await;
			if state.closed {
				return Err(transaction_coordinator_closed_error());
			}
			if self.transaction_coordinator.epoch.load(Ordering::Acquire) != epoch {
				return Err(transaction_expired_error(
					state
						.last_expired_timeout
						.unwrap_or(DEFAULT_TRANSACTION_TIMEOUT),
				));
			}
			if let Some(error) = transaction_known_state_error(&state, &key) {
				return Err(error);
			}
		}

		let (_, remote_session) = self
			.execute_backend_in_session("BEGIN".to_owned(), None, None)
			.await
			.map_err(map_transaction_connection_error)?;
		// A successful BEGIN response and the coordinator state update are two
		// separate async events. If the socket disconnected in that narrow gap,
		// pegboard-envoy has already dropped the connection-owned database handle
		// and rolled the transaction back. Never publish a handle for that stale
		// transaction.
		if let Some(session) = remote_session
			&& self.handle()?.connection_session() != Some(session)
		{
			return Err(transaction_connection_lost_error());
		}
		let operation = Arc::new(AsyncMutex::new(()));
		{
			let mut state = self.transaction_coordinator.state.lock().await;
			if state.closed {
				drop(state);
				if let Err(error) = self
					.execute_backend_in_session("ROLLBACK".to_owned(), None, remote_session)
					.await
				{
					tracing::error!(%error, "sqlite rollback after begin raced shutdown failed");
				}
				return Err(transaction_coordinator_closed_error());
			}
			state.active = Some(ActiveTransaction {
				key: key.clone(),
				timeout,
				expiring: false,
				connection_lost: false,
				remote_session,
				gate_guard: Some(gate_guard),
				operation,
				timeout_task: None,
			});
		}
		drop(permit);

		let timeout_task = CancellationToken::new();
		let mut state = self.transaction_coordinator.state.lock().await;
		if let Some(active) = state.active.as_mut().filter(|active| active.key == key) {
			active.timeout_task = Some(timeout_task.clone());
		} else {
			timeout_task.cancel();
		}
		drop(state);
		spawn_transaction_timeout(
			self.clone(),
			key.clone(),
			timeout,
			timeout_task,
			remote_session,
		);

		Ok(SqliteTransaction {
			db: self.clone(),
			key,
		})
	}

	async fn transaction_operation(&self, key: &str) -> Result<(Arc<AsyncMutex<()>>, Option<u64>)> {
		let state = self.transaction_coordinator.state.lock().await;
		if state.closed {
			return Err(transaction_coordinator_closed_error());
		}
		if let Some(active) = state.active.as_ref().filter(|active| active.key == key) {
			if active.expiring {
				if active.connection_lost {
					return Err(transaction_connection_lost_error());
				}
				return Err(transaction_expired_owner_error(key, active.timeout));
			}
			return Ok((Arc::clone(&active.operation), active.remote_session));
		}
		if let Some(error) = transaction_known_state_error(&state, key) {
			return Err(error);
		}
		Err(transaction_unknown_error(key))
	}

	async fn ensure_active_transaction(&self, key: &str) -> Result<()> {
		let state = self.transaction_coordinator.state.lock().await;
		if state.closed {
			return Err(transaction_coordinator_closed_error());
		}
		if let Some(active) = state.active.as_ref().filter(|active| active.key == key) {
			if active.expiring {
				if active.connection_lost {
					return Err(transaction_connection_lost_error());
				}
				return Err(transaction_expired_owner_error(key, active.timeout));
			}
			return Ok(());
		}
		if let Some(error) = transaction_known_state_error(&state, key) {
			return Err(error);
		}
		Err(transaction_unknown_error(key))
	}

	async fn transaction_exec(&self, key: &str, sql: String) -> Result<QueryResult> {
		let db = self.clone();
		let key = key.to_owned();
		run_detached_transaction_task(
			async move { db.transaction_exec_inner(&key, sql).await },
			"sqlite transaction exec task failed",
		)
		.await
	}

	async fn transaction_exec_inner(&self, key: &str, sql: String) -> Result<QueryResult> {
		let (operation, remote_session) = self.transaction_operation(key).await?;
		let _operation = operation.lock().await;
		self.ensure_active_transaction(key).await?;
		match self.exec_backend_in_session(sql, remote_session).await {
			Ok((result, _)) => Ok(result),
			Err(error) => Err(self.handle_transaction_backend_error(key, error).await),
		}
	}

	async fn transaction_execute(
		&self,
		key: &str,
		sql: String,
		params: Option<Vec<BindParam>>,
	) -> Result<ExecuteResult> {
		let db = self.clone();
		let key = key.to_owned();
		run_detached_transaction_task(
			async move { db.transaction_execute_inner(&key, sql, params).await },
			"sqlite transaction execute task failed",
		)
		.await
	}

	async fn transaction_execute_inner(
		&self,
		key: &str,
		sql: String,
		params: Option<Vec<BindParam>>,
	) -> Result<ExecuteResult> {
		let (operation, remote_session) = self.transaction_operation(key).await?;
		let _operation = operation.lock().await;
		self.ensure_active_transaction(key).await?;
		match self
			.execute_backend_in_session(sql, params, remote_session)
			.await
		{
			Ok((result, _)) => Ok(result),
			Err(error) => Err(self.handle_transaction_backend_error(key, error).await),
		}
	}

	async fn finish_transaction(&self, key: &str, commit: bool) -> Result<()> {
		let db = self.clone();
		let key = key.to_owned();
		run_detached_transaction_task(
			async move { db.finish_transaction_inner(&key, commit).await },
			"sqlite transaction finish task failed",
		)
		.await
	}

	async fn finish_transaction_inner(&self, key: &str, commit: bool) -> Result<()> {
		let (operation, remote_session) = self.transaction_operation(key).await?;
		let _operation = operation.lock().await;
		self.ensure_active_transaction(key).await?;

		let statement = if commit { "COMMIT" } else { "ROLLBACK" };
		let result = self
			.execute_backend_in_session(statement.to_owned(), None, remote_session)
			.await;
		if let Err(primary) = result {
			if is_remote_connection_error(&primary) {
				self.release_transaction(key, TransactionTerminalState::ConnectionLost, false)
					.await;
				return Err(self.attach_actor(map_transaction_connection_error(primary)));
			}
			if commit {
				let rollback_result = self
					.execute_backend_in_session("ROLLBACK".to_owned(), None, remote_session)
					.await;
				let rollback_failed = rollback_result
					.as_ref()
					.is_err_and(|error| !is_no_active_transaction_error(error));
				self.release_transaction(
					key,
					TransactionTerminalState::RolledBack,
					rollback_failed,
				)
				.await;
				if let Err(rollback) = rollback_result
					&& rollback_failed
				{
					tracing::error!(%rollback, "sqlite rollback after failed commit failed");
				}
				return Err(self.attach_actor(primary));
			}
			if is_no_active_transaction_error(&primary) {
				self.release_transaction(key, TransactionTerminalState::RolledBack, false)
					.await;
				return Ok(());
			}

			self.release_transaction(key, TransactionTerminalState::RolledBack, true)
				.await;
			return Err(self.attach_actor(primary));
		}

		self.release_transaction(
			key,
			if commit {
				TransactionTerminalState::Committed
			} else {
				TransactionTerminalState::RolledBack
			},
			false,
		)
		.await;
		Ok(())
	}

	async fn expire_transaction(&self, key: &str) -> Result<()> {
		let db = self.clone();
		let key = key.to_owned();
		run_detached_transaction_task(
			async move { db.expire_transaction_inner(&key).await },
			"sqlite transaction expiry task failed",
		)
		.await
	}

	async fn expire_transaction_inner(&self, key: &str) -> Result<()> {
		let (operation, timeout, remote_session) = {
			let mut state = self.transaction_coordinator.state.lock().await;
			if state.closed {
				return Err(transaction_coordinator_closed_error());
			}
			if let Some(active) = state.active.as_mut().filter(|active| active.key == key) {
				active.expiring = true;
				(
					Arc::clone(&active.operation),
					active.timeout,
					active.remote_session,
				)
			} else if let Some(error) = transaction_known_state_error(&state, key) {
				return Err(error);
			} else {
				return Err(transaction_unknown_error(key));
			}
		};
		let _operation = operation.lock().await;
		{
			let state = self.transaction_coordinator.state.lock().await;
			if state.closed {
				return Err(transaction_coordinator_closed_error());
			}
			if !state
				.active
				.as_ref()
				.is_some_and(|active| active.key == key)
			{
				if let Some(error) = transaction_known_state_error(&state, key) {
					return Err(error);
				}
				return Err(transaction_unknown_error(key));
			}
		}
		// The deadline only wins after any operation that already owned the
		// transaction settles. A commit or rollback that acquired the operation
		// lock first is allowed to finish without poisoning parked work.
		self.transaction_coordinator
			.epoch
			.fetch_add(1, Ordering::AcqRel);
		let rollback = self
			.execute_backend_in_session("ROLLBACK".to_owned(), None, remote_session)
			.await;
		let rollback_failed = rollback.as_ref().is_err_and(|error| {
			!is_no_active_transaction_error(error) && !is_remote_connection_error(error)
		});
		// Advance again before releasing the exclusive gate. Calls submitted both
		// before and during expiry cleanup must observe a changed epoch and fail;
		// only calls submitted after cleanup may proceed.
		self.transaction_coordinator
			.epoch
			.fetch_add(1, Ordering::AcqRel);
		self.release_transaction(
			key,
			TransactionTerminalState::Expired(timeout),
			rollback_failed,
		)
		.await;
		if let Err(error) = rollback
			&& rollback_failed
		{
			return Err(self.attach_actor(error));
		}
		Ok(())
	}

	async fn connection_lost_transaction(&self, key: &str, expected_session: u64) -> Result<()> {
		let (operation, still_expected) = {
			let mut state = self.transaction_coordinator.state.lock().await;
			let Some(active) = state.active.as_mut().filter(|active| active.key == key) else {
				return Ok(());
			};
			let still_expected = active.remote_session == Some(expected_session);
			if still_expected {
				active.expiring = true;
				active.connection_lost = true;
			}
			(Arc::clone(&active.operation), still_expected)
		};
		if !still_expected {
			return Ok(());
		}
		let _operation = operation.lock().await;

		// The remote database handle is scoped to the server-side WebSocket
		// connection. Disconnect closes it and SQLite rolls back its open
		// transaction. Sending ROLLBACK on a replacement connection would target a
		// different database session, so cleanup here is intentionally local: make
		// the handle terminal and release actor/socket work queued behind it.
		self.release_transaction(key, TransactionTerminalState::ConnectionLost, false)
			.await;
		Ok(())
	}

	async fn handle_transaction_backend_error(
		&self,
		key: &str,
		error: anyhow::Error,
	) -> anyhow::Error {
		if is_remote_connection_error(&error) {
			self.release_transaction(key, TransactionTerminalState::ConnectionLost, false)
				.await;
		}
		self.attach_actor(map_transaction_connection_error(error))
	}

	pub(super) async fn release_transaction(
		&self,
		key: &str,
		terminal: TransactionTerminalState,
		close: bool,
	) {
		let mut state = self.transaction_coordinator.state.lock().await;
		if !state
			.active
			.as_ref()
			.is_some_and(|active| active.key == key)
		{
			return;
		}
		let mut active = state.active.take().expect("active transaction was checked");
		if let Some(task) = active.timeout_task.take() {
			task.cancel();
		}
		if let TransactionTerminalState::Expired(timeout) = terminal {
			state.last_expired_timeout = Some(timeout);
		}
		insert_terminal_state(&mut state, key.to_owned(), terminal);
		state.closed |= close;
		if close {
			self.transaction_coordinator.admission.close();
			if let Ok(config) = self.runtime_config() {
				report_sqlite_worker_fatal(
					&self.worker_fatal_reported,
					config,
					"sqlite transaction rollback failed; coordinator closed".to_owned(),
				);
			}
		}
		drop(active.gate_guard.take());
	}

	pub(super) async fn shutdown_transaction_coordinator(&self) -> OwnedRwLockWriteGuard<()> {
		let active_operation = {
			let mut state = self.transaction_coordinator.state.lock().await;
			state.closed = true;
			state.active.as_ref().map(|active| {
				(
					active.key.clone(),
					Arc::clone(&active.operation),
					active.remote_session,
				)
			})
		};
		self.transaction_coordinator.admission.close();
		if let Some((key, operation, remote_session)) = active_operation {
			let _operation = operation.lock().await;
			let still_active = {
				let state = self.transaction_coordinator.state.lock().await;
				state
					.active
					.as_ref()
					.is_some_and(|active| active.key == key)
			};
			if still_active {
				let rollback = self
					.execute_backend_in_session("ROLLBACK".to_owned(), None, remote_session)
					.await;
				let rollback_failed = rollback.as_ref().is_err_and(|error| {
					!is_no_active_transaction_error(error) && !is_remote_connection_error(error)
				});
				if let Err(error) = &rollback
					&& rollback_failed
				{
					tracing::error!(%error, "sqlite rollback during coordinator shutdown failed");
				}
				self.transaction_coordinator
					.epoch
					.fetch_add(1, Ordering::AcqRel);
				self.release_transaction(
					&key,
					TransactionTerminalState::RolledBack,
					rollback_failed,
				)
				.await;
			}
		}
		Arc::clone(&self.transaction_coordinator.gate)
			.write_owned()
			.await
	}
}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_queue_full",
	"SQLite transaction queue is full.",
	"SQLite transaction coordinator queue is full. Limit is 128 operations."
)]
pub struct TransactionQueueFullError;

impl fmt::Display for TransactionQueueFullError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.write_str("sqlite transaction coordinator queue is full")
	}
}

impl Error for TransactionQueueFullError {}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_closed",
	"SQLite transaction coordinator is closed."
)]
pub struct TransactionCoordinatorClosedError;

impl fmt::Display for TransactionCoordinatorClosedError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.write_str("sqlite transaction coordinator is closed")
	}
}

impl Error for TransactionCoordinatorClosedError {}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_invalid_argument",
	"Invalid SQLite transaction argument.",
	"{message}"
)]
pub struct TransactionInvalidArgumentError {
	pub message: &'static str,
}

impl fmt::Display for TransactionInvalidArgumentError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.write_str(self.message)
	}
}

impl Error for TransactionInvalidArgumentError {}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_connection_lost",
	"SQLite transaction connection was lost.",
	"The Envoy connection that owned this SQLite transaction disconnected. The transaction was rolled back and cannot be resumed; start a new db.transaction()."
)]
pub struct TransactionConnectionLostError;

impl fmt::Display for TransactionConnectionLostError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.write_str(
			"the Envoy connection that owned this SQLite transaction disconnected; the transaction was rolled back and cannot be resumed",
		)
	}
}

impl Error for TransactionConnectionLostError {}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_unknown",
	"Unknown SQLite transaction handle.",
	"Unknown SQLite transaction handle `{key}`."
)]
pub struct TransactionUnknownError {
	pub key: String,
}

impl fmt::Display for TransactionUnknownError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(f, "unknown sqlite transaction handle `{}`", self.key)
	}
}

impl Error for TransactionUnknownError {}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_terminal",
	"SQLite transaction handle is terminal.",
	"SQLite transaction handle `{key}` is already {state}."
)]
pub struct TransactionTerminalError {
	pub key: String,
	pub state: &'static str,
}

impl fmt::Display for TransactionTerminalError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(
			f,
			"sqlite transaction handle `{}` is already {}",
			self.key, self.state
		)
	}
}

impl Error for TransactionTerminalError {}

#[derive(rivet_error::RivetError, Debug, Serialize)]
#[error(
	"sqlite",
	"transaction_expired",
	"SQLite transaction expired.",
	"SQLite transaction expired after {timeout_ms} ms and was rolled back; this timeout is a deadlock-safety backstop. Increase the db.transaction() `timeout` option if the transaction legitimately needs longer, and check for a nested transaction or use of the outer `db` instead of `tx` inside the callback."
)]
pub struct TransactionExpiredError {
	pub timeout_ms: u64,
}

impl fmt::Display for TransactionExpiredError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(
			f,
			"sqlite transaction expired after {} ms and was rolled back; this timeout is a deadlock-safety backstop. Increase the db.transaction() `timeout` option if the transaction legitimately needs longer, and check for a nested transaction or use of the outer `db` instead of `tx` inside the callback",
			self.timeout_ms
		)
	}
}

impl Error for TransactionExpiredError {}

impl TransactionExpiredError {
	fn owner(_key: &str, timeout: Duration) -> Self {
		Self {
			timeout_ms: duration_millis(timeout),
		}
	}

	fn parked(timeout: Duration) -> Self {
		Self {
			timeout_ms: duration_millis(timeout),
		}
	}
}

fn duration_millis(duration: Duration) -> u64 {
	duration.as_millis().try_into().unwrap_or(u64::MAX)
}

async fn sleep_transaction_timeout(mut timeout: Duration) {
	// Browser timers commonly cap one wait at signed 32-bit milliseconds.
	// Chunking preserves the caller's unbounded product-level timeout on wasm.
	const MAX_TIMER_WAIT: Duration = Duration::from_millis(2_000_000_000);
	while timeout > MAX_TIMER_WAIT {
		crate::time::sleep(MAX_TIMER_WAIT).await;
		timeout = timeout.saturating_sub(MAX_TIMER_WAIT);
	}
	crate::time::sleep(timeout).await;
}

fn spawn_transaction_timeout(
	db: SqliteDb,
	key: String,
	timeout: Duration,
	cancel: CancellationToken,
	remote_session: Option<u64>,
) {
	let task = async move {
		let connection_lost = async {
			let Some(expected) = remote_session else {
				std::future::pending::<()>().await;
				return;
			};
			let Ok(handle) = db.handle() else {
				return;
			};
			let mut sessions = handle.subscribe_connection_session();
			loop {
				if *sessions.borrow_and_update() != expected {
					return;
				}
				if sessions.changed().await.is_err() {
					return;
				}
			}
		};
		tokio::pin!(connection_lost);

		let expired = tokio::select! {
			_ = cancel.cancelled() => return,
			_ = sleep_transaction_timeout(timeout) => true,
			_ = &mut connection_lost => false,
		};
		let result = if expired {
			db.expire_transaction(&key).await
		} else {
			// This covers all disconnect timing windows. During BEGIN there is no
			// active coordinator entry yet and the BEGIN caller receives its own
			// indeterminate/session error. Between statements this branch is the only
			// signal, and it prevents the next statement from autocommitting after a
			// fast reconnect. During a statement or COMMIT, request tracking preserves
			// the operation's indeterminate result while this terminalizes the handle.
			db.connection_lost_transaction(&key, remote_session.expect("checked above"))
				.await
		};
		if let Err(error) = result {
			if error.downcast_ref::<TransactionTerminalError>().is_none() {
				tracing::error!(%error, "sqlite transaction terminal cleanup failed");
			}
		}
	};

	#[cfg(target_arch = "wasm32")]
	wasm_bindgen_futures::spawn_local(task);

	#[cfg(not(target_arch = "wasm32"))]
	RuntimeSpawner::spawn(task);
}

#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_detached_transaction_task<F, T>(
	future: F,
	context: &'static str,
) -> Result<T>
where
	F: Future<Output = Result<T>> + Send + 'static,
	T: Send + 'static,
{
	RuntimeSpawner::spawn(future).await.context(context)?
}

#[cfg(target_arch = "wasm32")]
pub(super) async fn run_detached_transaction_task<F, T>(
	future: F,
	context: &'static str,
) -> Result<T>
where
	F: Future<Output = Result<T>> + 'static,
	T: 'static,
{
	let (response_tx, response_rx) = oneshot::channel();
	wasm_bindgen_futures::spawn_local(async move {
		let _ = response_tx.send(future.await);
	});
	response_rx.await.context(context)?
}

pub(super) fn insert_terminal_state(
	state: &mut TransactionCoordinatorState,
	key: String,
	terminal: TransactionTerminalState,
) {
	if let TransactionTerminalState::Expired(timeout) = terminal {
		state.poisoned.insert(key, timeout);
		return;
	}
	if state.terminal.insert(key.clone(), terminal).is_none() {
		state.terminal_order.push_back(key);
	}
	while state.terminal_order.len() > TRANSACTION_TERMINAL_CAPACITY {
		if let Some(expired_key) = state.terminal_order.pop_front() {
			state.terminal.remove(&expired_key);
		}
	}
}

fn transaction_known_state_error(
	state: &TransactionCoordinatorState,
	key: &str,
) -> Option<anyhow::Error> {
	if let Some(timeout) = state.poisoned.get(key).copied() {
		return Some(transaction_expired_owner_error(key, timeout));
	}
	state
		.terminal
		.get(key)
		.copied()
		.map(|terminal| transaction_terminal_error(key, terminal))
}

fn is_no_active_transaction_error(error: &anyhow::Error) -> bool {
	// SQLite does not expose a dedicated result code for "cannot commit/rollback
	// - no transaction is active"; both native SQLite and the remote executor
	// surface SQLITE_ERROR plus this stable SQLite-generated text. Keep the
	// classifier isolated here. In particular, ON CONFLICT ROLLBACK can end a
	// transaction before our cleanup call, and treating that cleanup response as
	// fatal would permanently close the coordinator. The cross-runtime actor-db
	// suite exercises that real automatic-rollback path on native and Wasm.
	error.chain().any(|cause| {
		cause
			.to_string()
			.to_ascii_lowercase()
			.contains("no transaction is active")
	})
}

fn transaction_queue_full_error() -> anyhow::Error {
	TransactionQueueFullError
		.build()
		.context(TransactionQueueFullError)
}

fn transaction_coordinator_closed_error() -> anyhow::Error {
	TransactionCoordinatorClosedError
		.build()
		.context(TransactionCoordinatorClosedError)
}

fn transaction_unknown_error(key: &str) -> anyhow::Error {
	TransactionUnknownError {
		key: key.to_owned(),
	}
	.build()
	.context(TransactionUnknownError {
		key: key.to_owned(),
	})
}

fn transaction_expired_error(timeout: Duration) -> anyhow::Error {
	TransactionExpiredError::parked(timeout)
		.build()
		.context(TransactionExpiredError::parked(timeout))
}

fn transaction_expired_owner_error(key: &str, timeout: Duration) -> anyhow::Error {
	TransactionExpiredError::owner(key, timeout)
		.build()
		.context(TransactionExpiredError::owner(key, timeout))
}

fn transaction_terminal_error(key: &str, terminal: TransactionTerminalState) -> anyhow::Error {
	match terminal {
		TransactionTerminalState::Expired(timeout) => transaction_expired_owner_error(key, timeout),
		TransactionTerminalState::Committed => transaction_terminal_state_error(key, "committed"),
		TransactionTerminalState::RolledBack => {
			transaction_terminal_state_error(key, "rolled back")
		}
		TransactionTerminalState::ConnectionLost => transaction_connection_lost_error(),
	}
}

fn transaction_invalid_argument_error(message: &'static str) -> anyhow::Error {
	TransactionInvalidArgumentError { message }
		.build()
		.context(TransactionInvalidArgumentError { message })
}

fn transaction_connection_lost_error() -> anyhow::Error {
	TransactionConnectionLostError
		.build()
		.context(TransactionConnectionLostError)
}

fn is_remote_connection_error(error: &anyhow::Error) -> bool {
	error
		.downcast_ref::<super::RemoteSqliteConnectionSessionLostError>()
		.is_some()
		|| error
			.downcast_ref::<super::RemoteSqliteIndeterminateResultError>()
			.is_some()
}

fn map_transaction_connection_error(error: anyhow::Error) -> anyhow::Error {
	if error
		.downcast_ref::<super::RemoteSqliteConnectionSessionLostError>()
		.is_some()
	{
		return transaction_connection_lost_error();
	}
	super::remote_request_error(error)
}

fn transaction_terminal_state_error(key: &str, state: &'static str) -> anyhow::Error {
	TransactionTerminalError {
		key: key.to_owned(),
		state,
	}
	.build()
	.context(TransactionTerminalError {
		key: key.to_owned(),
		state,
	})
}