use std::collections::BTreeMap;
use std::future::Future;
use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
use std::sync::{Arc, Mutex, OnceLock, Weak};
pub const DEFAULT_DATASOURCE: &str = "default";
pub mod observability;
pub const MAX_DATASOURCE_NAME_BYTES: usize = 128;
pub const MAX_MANAGED_DATASOURCES: usize = 64;
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Deserialize, serde::Serialize,
)]
#[serde(rename_all = "lowercase")]
pub enum DatabaseDriver {
MySql,
PostgreSql,
Sqlite,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct DatabaseCapabilities(u8);
impl DatabaseCapabilities {
const MYSQL: u8 = 1;
const POSTGRESQL: u8 = 1 << 1;
const SQLITE: u8 = 1 << 2;
pub const fn empty() -> Self {
Self(0)
}
pub const fn with(self, driver: DatabaseDriver) -> Self {
let flag = match driver {
DatabaseDriver::MySql => Self::MYSQL,
DatabaseDriver::PostgreSql => Self::POSTGRESQL,
DatabaseDriver::Sqlite => Self::SQLITE,
};
Self(self.0 | flag)
}
pub const fn contains(self, driver: DatabaseDriver) -> bool {
let flag = match driver {
DatabaseDriver::MySql => Self::MYSQL,
DatabaseDriver::PostgreSql => Self::POSTGRESQL,
DatabaseDriver::Sqlite => Self::SQLITE,
};
self.0 & flag != 0
}
}
impl std::fmt::Display for DatabaseDriver {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(match self {
Self::MySql => "mysql",
Self::PostgreSql => "postgresql",
Self::Sqlite => "sqlite",
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum DatasourceNameError {
#[error("datasource name cannot be empty")]
Empty,
#[error("datasource name cannot contain leading or trailing whitespace")]
SurroundingWhitespace,
#[error("datasource name must use ASCII letters, digits, '.', '_' or '-' and stay within the byte limit")]
InvalidCharacters,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct DatasourceRef(Arc<str>);
impl DatasourceRef {
pub fn new(name: impl AsRef<str>) -> anyhow::Result<Self> {
Self::try_new(name).map_err(anyhow::Error::new)
}
pub fn try_new(name: impl AsRef<str>) -> Result<Self, DatasourceNameError> {
let name = name.as_ref();
validate_datasource_name(name)?;
Ok(Self(Arc::from(name)))
}
pub fn default_ref() -> Self {
Self(Arc::from(DEFAULT_DATASOURCE))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl Default for DatasourceRef {
fn default() -> Self {
Self::default_ref()
}
}
impl AsRef<str> for DatasourceRef {
fn as_ref(&self) -> &str {
self.as_str()
}
}
impl std::fmt::Display for DatasourceRef {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(self.as_str())
}
}
pub fn validate_datasource_name(name: &str) -> Result<(), DatasourceNameError> {
if name.is_empty() {
return Err(DatasourceNameError::Empty);
}
if name.trim() != name {
return Err(DatasourceNameError::SurroundingWhitespace);
}
if name.len() > MAX_DATASOURCE_NAME_BYTES
|| !name
.bytes()
.all(|value| value.is_ascii_alphanumeric() || matches!(value, b'.' | b'_' | b'-'))
{
return Err(DatasourceNameError::InvalidCharacters);
}
Ok(())
}
#[derive(Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DataSourcePoolConfig {
pub url: String,
#[serde(default = "default_max_connections")]
pub max_connections: u32,
#[serde(default)]
pub min_connections: u32,
#[serde(default = "default_acquire_timeout_ms")]
pub acquire_timeout_ms: u64,
#[serde(default = "default_connect_timeout_ms")]
pub connect_timeout_ms: u64,
#[serde(default = "default_probe_on_start")]
pub probe_on_start: bool,
}
impl std::fmt::Debug for DataSourcePoolConfig {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("DataSourcePoolConfig")
.field("url", &redact_url(&self.url))
.field("max_connections", &self.max_connections)
.field("min_connections", &self.min_connections)
.field("acquire_timeout_ms", &self.acquire_timeout_ms)
.field("connect_timeout_ms", &self.connect_timeout_ms)
.field("probe_on_start", &self.probe_on_start)
.finish()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum PoolConfigError {
#[error("datasource url cannot be empty")]
EmptyUrl,
#[error("datasource max_connections must be greater than zero")]
ZeroMaxConnections,
#[error("datasource min_connections cannot exceed max_connections")]
InvalidConnectionRange,
#[error("datasource acquire_timeout_ms must be greater than zero")]
ZeroAcquireTimeout,
#[error("datasource connect_timeout_ms must be greater than zero")]
ZeroConnectTimeout,
}
impl DataSourcePoolConfig {
pub fn validate_common(&self) -> Result<(), PoolConfigError> {
if self.url.trim().is_empty() {
return Err(PoolConfigError::EmptyUrl);
}
if self.max_connections == 0 {
return Err(PoolConfigError::ZeroMaxConnections);
}
if self.min_connections > self.max_connections {
return Err(PoolConfigError::InvalidConnectionRange);
}
if self.acquire_timeout_ms == 0 {
return Err(PoolConfigError::ZeroAcquireTimeout);
}
if self.connect_timeout_ms == 0 {
return Err(PoolConfigError::ZeroConnectTimeout);
}
Ok(())
}
pub fn endpoint(&self) -> String {
redacted_endpoint(&self.url)
}
}
pub fn redact_url(url: &str) -> String {
let Some((scheme, rest)) = url.split_once("://") else {
return "<redacted-url>".to_owned();
};
if scheme.is_empty()
|| !scheme
.bytes()
.all(|value| value.is_ascii_alphanumeric() || matches!(value, b'+' | b'-' | b'.'))
{
return "<redacted-url>".to_owned();
}
let endpoint = redacted_endpoint(url);
if endpoint == "<unknown-endpoint>" {
return "<redacted-url>".to_owned();
}
let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
if authority.contains('@') {
format!("{scheme}://***@{endpoint}")
} else {
format!("{scheme}://{endpoint}")
}
}
pub fn redacted_endpoint(url: &str) -> String {
let Some((_, rest)) = url.split_once("://") else {
return "<unknown-endpoint>".to_owned();
};
let authority_and_path = rest.rsplit_once('@').map_or(rest, |(_, value)| value);
let endpoint = authority_and_path
.split(['?', '#'])
.next()
.unwrap_or(authority_and_path);
if endpoint.is_empty() {
"<unknown-endpoint>".to_owned()
} else {
endpoint.to_owned()
}
}
fn default_max_connections() -> u32 {
10
}
fn default_acquire_timeout_ms() -> u64 {
2_000
}
fn default_connect_timeout_ms() -> u64 {
5_000
}
fn default_probe_on_start() -> bool {
true
}
static NEXT_OWNER_ID: AtomicU64 = AtomicU64::new(1);
#[derive(Debug)]
struct OwnerIdentity {
id: u64,
}
#[derive(Clone)]
pub struct ManagedRegistryOwner(Arc<OwnerIdentity>);
impl ManagedRegistryOwner {
pub fn new() -> Self {
Self(Arc::new(OwnerIdentity {
id: NEXT_OWNER_ID.fetch_add(1, Ordering::Relaxed),
}))
}
pub fn ptr_eq(&self, other: &Self) -> bool {
Arc::ptr_eq(&self.0, &other.0)
}
pub fn diagnostic_id(&self) -> u64 {
self.0.id
}
fn downgrade(&self) -> Weak<OwnerIdentity> {
Arc::downgrade(&self.0)
}
}
impl Default for ManagedRegistryOwner {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Debug for ManagedRegistryOwner {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_tuple("ManagedRegistryOwner")
.field(&self.diagnostic_id())
.finish()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BootstrapKind {
Configured,
DeferredDefault,
}
#[derive(Clone, Debug)]
pub struct ManagedInstallationToken {
owner: ManagedRegistryOwner,
kind: BootstrapKind,
}
impl ManagedInstallationToken {
pub fn owner(&self) -> &ManagedRegistryOwner {
&self.owner
}
pub fn kind(&self) -> BootstrapKind {
self.kind
}
}
pub struct DataSourceCatalog {
owner: ManagedRegistryOwner,
entries: BTreeMap<DatasourceRef, DatabaseDriver>,
lifecycle: AtomicU8,
}
const CATALOG_STAGING: u8 = 0;
const CATALOG_OPEN: u8 = 1;
const CATALOG_STOPPED: u8 = 2;
impl DataSourceCatalog {
pub fn try_new(
owner: ManagedRegistryOwner,
entries: impl IntoIterator<Item = (String, DatabaseDriver)>,
) -> Result<Self, CatalogBuildError> {
let mut catalog = BTreeMap::new();
for (name, driver) in entries {
if catalog.len() >= MAX_MANAGED_DATASOURCES {
return Err(CatalogBuildError::TooManyDatasources);
}
let reference = DatasourceRef::try_new(name).map_err(CatalogBuildError::InvalidName)?;
if catalog.insert(reference.clone(), driver).is_some() {
return Err(CatalogBuildError::Duplicate(reference));
}
}
if catalog.is_empty() {
return Err(CatalogBuildError::Empty);
}
Ok(Self {
owner,
entries: catalog,
lifecycle: AtomicU8::new(CATALOG_STAGING),
})
}
pub fn owner(&self) -> &ManagedRegistryOwner {
&self.owner
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn entries(&self) -> Vec<(DatasourceRef, DatabaseDriver)> {
self.entries
.iter()
.map(|(reference, driver)| (reference.clone(), *driver))
.collect()
}
pub fn resolve(
&self,
datasource: &str,
expected: DatabaseDriver,
) -> Result<DatasourceRef, DataSourceLookupError> {
match self.lifecycle.load(Ordering::Acquire) {
CATALOG_OPEN => {}
CATALOG_STOPPED => return Err(DataSourceLookupError::Stopped),
_ => return Err(DataSourceLookupError::RegistryUnavailable),
}
let reference =
DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)?;
let actual = self.entries.get(&reference).copied().ok_or_else(|| {
DataSourceLookupError::NotFound {
datasource: reference.clone(),
}
})?;
if actual != expected {
return Err(DataSourceLookupError::DriverMismatch {
datasource: reference,
expected,
actual,
});
}
Ok(reference)
}
fn open(&self) {
self.lifecycle.store(CATALOG_OPEN, Ordering::Release);
}
fn stop_accepting(&self) {
self.lifecycle.store(CATALOG_STOPPED, Ordering::Release);
}
}
#[derive(Debug, thiserror::Error)]
pub enum CatalogBuildError {
#[error("managed datasource catalog cannot be empty")]
Empty,
#[error("managed datasource count exceeds the supported limit")]
TooManyDatasources,
#[error("invalid datasource name")]
InvalidName(#[source] DatasourceNameError),
#[error("datasource `{0}` is configured more than once")]
Duplicate(DatasourceRef),
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum DataSourceLookupError {
#[error("invalid datasource name")]
InvalidName,
#[error("datasource registry is unavailable")]
RegistryUnavailable,
#[error("datasource registry is stopped")]
Stopped,
#[error("datasource `{datasource}` is not managed by this application")]
NotFound {
datasource: DatasourceRef,
},
#[error("datasource `{datasource}` uses {actual}, not {expected}")]
DriverMismatch {
datasource: DatasourceRef,
expected: DatabaseDriver,
actual: DatabaseDriver,
},
}
#[must_use]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TxDecision<T, E> {
Commit(T),
Rollback(E),
}
#[derive(Debug)]
pub enum TxRollbackCause<E> {
Decision(E),
RollbackOnly {
reason: String,
},
}
#[derive(Debug)]
pub enum TxRunError<E> {
Rollback(E),
RollbackOnly {
reason: String,
},
CommitRejected {
reason: String,
},
CommitUncertain {
reason: String,
},
RollbackFailed {
cause: TxRollbackCause<E>,
reason: String,
},
Infrastructure {
reason: String,
},
}
#[derive(Debug)]
pub enum TxEntryError<E> {
Lookup(DataSourceLookupError),
Run(TxRunError<E>),
}
impl<E: std::fmt::Display> std::fmt::Display for TxEntryError<E> {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Lookup(error) => {
write!(formatter, "transaction datasource lookup failed: {error}")
}
Self::Run(error) => std::fmt::Display::fmt(error, formatter),
}
}
}
impl<E> std::error::Error for TxEntryError<E> where E: std::fmt::Debug + std::fmt::Display {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CommitOutcome {
Committed,
Rejected,
OutcomeUnknown,
}
impl<E: std::fmt::Display> std::fmt::Display for TxRunError<E> {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Rollback(error) => write!(formatter, "transaction rolled back: {error}"),
Self::RollbackOnly {
reason,
} => {
write!(formatter, "transaction was rollback-only: {reason}")
}
Self::CommitRejected {
reason,
} => write!(formatter, "commit rejected: {reason}"),
Self::CommitUncertain {
reason,
} => write!(formatter, "commit uncertain: {reason}"),
Self::RollbackFailed {
reason,
..
} => write!(formatter, "rollback failed: {reason}"),
Self::Infrastructure {
reason,
} => {
write!(formatter, "transaction infrastructure failed: {reason}")
}
}
}
}
impl<E> std::error::Error for TxRunError<E> where E: std::fmt::Debug + std::fmt::Display {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("ambient transaction driver mismatch: active={active}, requested={requested}")]
pub struct DriverScopeError {
pub active: DatabaseDriver,
pub requested: DatabaseDriver,
}
tokio::task_local! {
static ACTIVE_DRIVER: DatabaseDriver;
}
pub fn current_driver() -> Option<DatabaseDriver> {
ACTIVE_DRIVER.try_with(|driver| *driver).ok()
}
pub fn ensure_driver(requested: DatabaseDriver) -> Result<(), DriverScopeError> {
match current_driver() {
Some(active) if active != requested => Err(DriverScopeError {
active,
requested,
}),
_ => Ok(()),
}
}
pub async fn scope_driver<F>(
driver: DatabaseDriver,
future: F,
) -> Result<F::Output, DriverScopeError>
where
F: Future,
{
match current_driver() {
Some(active) if active != driver => Err(DriverScopeError {
active,
requested: driver,
}),
Some(_) => Ok(future.await),
None => Ok(ACTIVE_DRIVER.scope(driver, future).await),
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum RegistryCoordinationError {
#[error("standalone datasource driver conflicts with the active driver")]
StandaloneDriverConflict,
#[error("standalone datasource registration conflicts with managed application state")]
ManagedModeConflict,
#[error("deferred datasource bootstrap accepts only the default datasource")]
DeferredOnlyDefault,
#[error("deferred datasource bootstrap already uses another driver")]
DeferredDriverConflict,
#[error("datasource registry is already occupied")]
RegistryOccupied,
#[error("managed registry owner token does not match the active bootstrap")]
OwnerMismatch,
#[error("datasource catalog owner does not match the installation token")]
CatalogOwnerMismatch,
#[error("datasource catalog identity does not match the active managed catalog")]
CatalogIdentityMismatch,
#[error("deferred datasource bootstrap did not install the expected default driver")]
DeferredDriverMissing,
#[error("deferred datasource resources must be removed before bootstrap cleanup")]
DeferredResourcesRemain,
#[error("public standalone adopt is unavailable during application bootstrap")]
PublicAdoptDuringBootstrap,
#[error("orphaned deferred datasource resources require explicit cleanup")]
OrphanedDeferredResources,
#[error("deferred datasource catalog must contain only the installed default driver")]
InvalidDeferredCatalog,
}
#[derive(Debug)]
enum RegistryMode {
Empty,
Standalone(DatabaseDriver),
Bootstrapping {
owner: Weak<OwnerIdentity>,
kind: BootstrapKind,
deferred_driver: Option<DatabaseDriver>,
},
Managed {
owner: Weak<OwnerIdentity>,
catalog: Weak<DataSourceCatalog>,
},
}
#[derive(Debug)]
struct RegistryState {
mode: RegistryMode,
}
impl Default for RegistryState {
fn default() -> Self {
Self {
mode: RegistryMode::Empty,
}
}
}
static REGISTRY_STATE: OnceLock<Mutex<RegistryState>> = OnceLock::new();
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RegistryModeSnapshot {
Empty,
Standalone(DatabaseDriver),
Bootstrapping {
kind: BootstrapKind,
deferred_driver: Option<DatabaseDriver>,
},
Managed,
OrphanedDeferred(DatabaseDriver),
}
#[doc(hidden)]
pub fn coordinate<T>(operation: impl FnOnce(&mut RegistryCoordinator<'_>) -> T) -> T {
let mut state = REGISTRY_STATE
.get_or_init(|| Mutex::new(RegistryState::default()))
.lock()
.unwrap_or_else(|error| error.into_inner());
let mut coordinator = RegistryCoordinator {
state: &mut state,
};
coordinator.sanitize();
operation(&mut coordinator)
}
#[doc(hidden)]
pub struct RegistryCoordinator<'a> {
state: &'a mut RegistryState,
}
impl RegistryCoordinator<'_> {
fn sanitize(&mut self) {
match &self.state.mode {
RegistryMode::Bootstrapping {
owner,
deferred_driver: None,
..
} if owner.upgrade().is_none() => {
self.state.mode = RegistryMode::Empty;
}
RegistryMode::Managed {
owner,
catalog,
} if owner.upgrade().is_none() && catalog.upgrade().is_none() => {
self.state.mode = RegistryMode::Empty;
}
_ => {}
}
}
pub fn register_standalone(
&mut self,
driver: DatabaseDriver,
datasource: &str,
) -> Result<(), RegistryCoordinationError> {
match &mut self.state.mode {
RegistryMode::Empty => {
self.state.mode = RegistryMode::Standalone(driver);
Ok(())
}
RegistryMode::Standalone(active) if *active == driver => Ok(()),
RegistryMode::Standalone(_) => Err(RegistryCoordinationError::StandaloneDriverConflict),
RegistryMode::Bootstrapping {
owner,
kind: BootstrapKind::DeferredDefault,
deferred_driver,
} => {
if owner.upgrade().is_none() && deferred_driver.is_some() {
return Err(RegistryCoordinationError::OrphanedDeferredResources);
}
if datasource != DEFAULT_DATASOURCE {
return Err(RegistryCoordinationError::DeferredOnlyDefault);
}
match deferred_driver {
Some(active) if *active != driver => {
Err(RegistryCoordinationError::DeferredDriverConflict)
}
Some(_) => Ok(()),
slot @ None => {
*slot = Some(driver);
Ok(())
}
}
}
RegistryMode::Bootstrapping {
..
}
| RegistryMode::Managed {
..
} => Err(RegistryCoordinationError::ManagedModeConflict),
}
}
pub fn release_standalone(
&mut self,
driver: DatabaseDriver,
) -> Result<(), RegistryCoordinationError> {
match &mut self.state.mode {
RegistryMode::Standalone(active) if *active == driver => {
self.state.mode = RegistryMode::Empty;
Ok(())
}
RegistryMode::Bootstrapping {
owner,
kind: BootstrapKind::DeferredDefault,
deferred_driver: Some(active),
} if *active == driver => {
let owner_alive = owner.upgrade().is_some();
if owner_alive {
if let RegistryMode::Bootstrapping {
deferred_driver,
..
} = &mut self.state.mode
{
*deferred_driver = None;
}
} else {
self.state.mode = RegistryMode::Empty;
}
Ok(())
}
RegistryMode::Empty => Ok(()),
_ => Err(RegistryCoordinationError::ManagedModeConflict),
}
}
pub fn begin_bootstrap(
&mut self,
owner: &ManagedRegistryOwner,
kind: BootstrapKind,
) -> Result<ManagedInstallationToken, RegistryCoordinationError> {
if !matches!(self.state.mode, RegistryMode::Empty) {
return Err(RegistryCoordinationError::RegistryOccupied);
}
self.state.mode = RegistryMode::Bootstrapping {
owner: owner.downgrade(),
kind,
deferred_driver: None,
};
Ok(ManagedInstallationToken {
owner: owner.clone(),
kind,
})
}
pub fn begin_legacy_adopt(
&mut self,
owner: &ManagedRegistryOwner,
driver: DatabaseDriver,
) -> Result<ManagedInstallationToken, RegistryCoordinationError> {
match &self.state.mode {
RegistryMode::Standalone(active) if *active == driver => {
self.state.mode = RegistryMode::Bootstrapping {
owner: owner.downgrade(),
kind: BootstrapKind::Configured,
deferred_driver: None,
};
Ok(ManagedInstallationToken {
owner: owner.clone(),
kind: BootstrapKind::Configured,
})
}
RegistryMode::Bootstrapping {
..
} => Err(RegistryCoordinationError::PublicAdoptDuringBootstrap),
_ => Err(RegistryCoordinationError::RegistryOccupied),
}
}
pub fn restore_legacy_standalone(
&mut self,
token: &ManagedInstallationToken,
driver: DatabaseDriver,
) -> Result<(), RegistryCoordinationError> {
self.verify_bootstrap_owner(token)?;
self.state.mode = RegistryMode::Standalone(driver);
Ok(())
}
pub fn verify_deferred_adopt(
&mut self,
token: &ManagedInstallationToken,
driver: DatabaseDriver,
) -> Result<(), RegistryCoordinationError> {
self.verify_bootstrap_owner(token)?;
match &self.state.mode {
RegistryMode::Bootstrapping {
kind: BootstrapKind::DeferredDefault,
deferred_driver: Some(active),
..
} if *active == driver => Ok(()),
_ => Err(RegistryCoordinationError::DeferredDriverMissing),
}
}
pub fn verify_installation_token(
&self,
token: &ManagedInstallationToken,
) -> Result<(), RegistryCoordinationError> {
self.verify_bootstrap_owner(token)
}
pub fn install_catalog(
&mut self,
token: &ManagedInstallationToken,
catalog: &Arc<DataSourceCatalog>,
) -> Result<(), RegistryCoordinationError> {
self.verify_bootstrap_owner(token)?;
if !catalog.owner().ptr_eq(token.owner()) {
return Err(RegistryCoordinationError::CatalogOwnerMismatch);
}
if let RegistryMode::Bootstrapping {
kind: BootstrapKind::DeferredDefault,
deferred_driver: Some(driver),
..
} = &self.state.mode
{
let entries = catalog.entries();
if entries.len() != 1
|| entries[0].0.as_str() != DEFAULT_DATASOURCE
|| entries[0].1 != *driver
{
return Err(RegistryCoordinationError::InvalidDeferredCatalog);
}
}
self.state.mode = RegistryMode::Managed {
owner: token.owner.downgrade(),
catalog: Arc::downgrade(catalog),
};
Ok(())
}
pub fn open_catalog(
&mut self,
token: &ManagedInstallationToken,
catalog: &Arc<DataSourceCatalog>,
) -> Result<(), RegistryCoordinationError> {
match &self.state.mode {
RegistryMode::Managed {
owner,
catalog: active,
} => {
let Some(active_owner) = owner.upgrade() else {
return Err(RegistryCoordinationError::OwnerMismatch);
};
if !Arc::ptr_eq(&active_owner, &token.owner.0) {
return Err(RegistryCoordinationError::OwnerMismatch);
}
let Some(active_catalog) = active.upgrade() else {
return Err(RegistryCoordinationError::CatalogIdentityMismatch);
};
if !Arc::ptr_eq(&active_catalog, catalog) {
return Err(RegistryCoordinationError::CatalogIdentityMismatch);
}
catalog.open();
Ok(())
}
_ => Err(RegistryCoordinationError::OwnerMismatch),
}
}
pub fn clear_managed(
&mut self,
owner: &ManagedRegistryOwner,
) -> Result<(), RegistryCoordinationError> {
match &self.state.mode {
RegistryMode::Managed {
owner: active,
catalog,
} => {
let Some(active_owner) = active.upgrade() else {
return Err(RegistryCoordinationError::OwnerMismatch);
};
if !Arc::ptr_eq(&active_owner, &owner.0) {
return Err(RegistryCoordinationError::OwnerMismatch);
}
if let Some(catalog) = catalog.upgrade() {
catalog.stop_accepting();
}
self.state.mode = RegistryMode::Empty;
Ok(())
}
_ => Err(RegistryCoordinationError::OwnerMismatch),
}
}
pub fn clear_deferred_driver(
&mut self,
token: &ManagedInstallationToken,
driver: DatabaseDriver,
) -> Result<(), RegistryCoordinationError> {
self.verify_bootstrap_owner(token)?;
match &mut self.state.mode {
RegistryMode::Bootstrapping {
kind: BootstrapKind::DeferredDefault,
deferred_driver: Some(active),
..
} if *active == driver => {
if let RegistryMode::Bootstrapping {
deferred_driver,
..
} = &mut self.state.mode
{
*deferred_driver = None;
}
Ok(())
}
_ => Err(RegistryCoordinationError::DeferredDriverMissing),
}
}
pub fn abort_bootstrap(
&mut self,
token: &ManagedInstallationToken,
) -> Result<(), RegistryCoordinationError> {
self.verify_bootstrap_owner(token)?;
match &self.state.mode {
RegistryMode::Bootstrapping {
deferred_driver: Some(_),
..
} => Err(RegistryCoordinationError::DeferredResourcesRemain),
RegistryMode::Bootstrapping {
..
} => {
self.state.mode = RegistryMode::Empty;
Ok(())
}
_ => Err(RegistryCoordinationError::OwnerMismatch),
}
}
pub fn snapshot(&self) -> RegistryModeSnapshot {
match &self.state.mode {
RegistryMode::Empty => RegistryModeSnapshot::Empty,
RegistryMode::Standalone(driver) => RegistryModeSnapshot::Standalone(*driver),
RegistryMode::Bootstrapping {
owner,
kind,
deferred_driver,
} => match (owner.upgrade(), deferred_driver) {
(None, Some(driver)) => RegistryModeSnapshot::OrphanedDeferred(*driver),
_ => RegistryModeSnapshot::Bootstrapping {
kind: *kind,
deferred_driver: *deferred_driver,
},
},
RegistryMode::Managed {
..
} => RegistryModeSnapshot::Managed,
}
}
fn verify_bootstrap_owner(
&self,
token: &ManagedInstallationToken,
) -> Result<(), RegistryCoordinationError> {
match &self.state.mode {
RegistryMode::Bootstrapping {
owner,
kind,
..
} => {
let Some(active) = owner.upgrade() else {
return Err(RegistryCoordinationError::OwnerMismatch);
};
if Arc::ptr_eq(&active, &token.owner.0) && *kind == token.kind {
Ok(())
} else {
Err(RegistryCoordinationError::OwnerMismatch)
}
}
_ => Err(RegistryCoordinationError::OwnerMismatch),
}
}
}
pub fn registry_mode() -> RegistryModeSnapshot {
coordinate(|coordinator| coordinator.snapshot())
}
pub fn begin_managed_bootstrap(
kind: BootstrapKind,
) -> Result<(ManagedRegistryOwner, ManagedInstallationToken), RegistryCoordinationError> {
let owner = ManagedRegistryOwner::new();
let token = coordinate(|coordinator| coordinator.begin_bootstrap(&owner, kind))?;
Ok((owner, token))
}
pub fn install_managed_catalog(
token: &ManagedInstallationToken,
catalog: &Arc<DataSourceCatalog>,
) -> Result<(), RegistryCoordinationError> {
coordinate(|coordinator| coordinator.install_catalog(token, catalog))
}
pub fn open_managed_catalog(
token: &ManagedInstallationToken,
catalog: &Arc<DataSourceCatalog>,
) -> Result<(), RegistryCoordinationError> {
coordinate(|coordinator| coordinator.open_catalog(token, catalog))
}
pub fn clear_managed_catalog(
owner: &ManagedRegistryOwner,
) -> Result<(), RegistryCoordinationError> {
coordinate(|coordinator| coordinator.clear_managed(owner))
}
pub fn abort_managed_bootstrap(
token: &ManagedInstallationToken,
) -> Result<(), RegistryCoordinationError> {
coordinate(|coordinator| coordinator.abort_bootstrap(token))
}
pub fn resolve_datasource(
datasource: &str,
expected: DatabaseDriver,
) -> Result<DatasourceRef, DataSourceLookupError> {
coordinate(|coordinator| match &coordinator.state.mode {
RegistryMode::Empty => {
DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
}
RegistryMode::Bootstrapping {
kind: BootstrapKind::DeferredDefault,
deferred_driver: Some(actual),
..
} if *actual == expected && datasource == DEFAULT_DATASOURCE => {
DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
}
RegistryMode::Bootstrapping {
..
} => Err(DataSourceLookupError::RegistryUnavailable),
RegistryMode::Standalone(actual) if *actual == expected => {
DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
}
RegistryMode::Standalone(actual) => {
let reference =
DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)?;
Err(DataSourceLookupError::DriverMismatch {
datasource: reference,
expected,
actual: *actual,
})
}
RegistryMode::Managed {
catalog,
..
} => catalog
.upgrade()
.ok_or(DataSourceLookupError::RegistryUnavailable)?
.resolve(datasource, expected),
})
}
pub fn managed_catalog() -> Result<Arc<DataSourceCatalog>, DataSourceLookupError> {
coordinate(|coordinator| match &coordinator.state.mode {
RegistryMode::Managed {
catalog,
..
} => catalog
.upgrade()
.ok_or(DataSourceLookupError::RegistryUnavailable)
.and_then(|catalog| match catalog.lifecycle.load(Ordering::Acquire) {
CATALOG_OPEN => Ok(catalog),
CATALOG_STOPPED => Err(DataSourceLookupError::Stopped),
_ => Err(DataSourceLookupError::RegistryUnavailable),
}),
_ => Err(DataSourceLookupError::RegistryUnavailable),
})
}