1use std::collections::BTreeMap;
11use std::future::Future;
12use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
13use std::sync::{Arc, Mutex, OnceLock, Weak};
14
15pub const DEFAULT_DATASOURCE: &str = "default";
17
18pub mod observability;
19pub const MAX_DATASOURCE_NAME_BYTES: usize = 128;
21pub const MAX_MANAGED_DATASOURCES: usize = 64;
23
24#[derive(
26 Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Deserialize, serde::Serialize,
27)]
28#[serde(rename_all = "lowercase")]
29pub enum DatabaseDriver {
30 MySql,
32 PostgreSql,
34 Sqlite,
36}
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
40pub struct DatabaseCapabilities(u8);
41
42impl DatabaseCapabilities {
43 const MYSQL: u8 = 1;
44 const POSTGRESQL: u8 = 1 << 1;
45 const SQLITE: u8 = 1 << 2;
46
47 pub const fn empty() -> Self {
53 Self(0)
54 }
55
56 pub const fn with(self, driver: DatabaseDriver) -> Self {
63 let flag = match driver {
64 DatabaseDriver::MySql => Self::MYSQL,
65 DatabaseDriver::PostgreSql => Self::POSTGRESQL,
66 DatabaseDriver::Sqlite => Self::SQLITE,
67 };
68 Self(self.0 | flag)
69 }
70
71 pub const fn contains(self, driver: DatabaseDriver) -> bool {
78 let flag = match driver {
79 DatabaseDriver::MySql => Self::MYSQL,
80 DatabaseDriver::PostgreSql => Self::POSTGRESQL,
81 DatabaseDriver::Sqlite => Self::SQLITE,
82 };
83 self.0 & flag != 0
84 }
85}
86
87impl std::fmt::Display for DatabaseDriver {
88 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
95 formatter.write_str(match self {
96 Self::MySql => "mysql",
97 Self::PostgreSql => "postgresql",
98 Self::Sqlite => "sqlite",
99 })
100 }
101}
102
103#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
105pub enum DatasourceNameError {
106 #[error("datasource name cannot be empty")]
108 Empty,
109 #[error("datasource name cannot contain leading or trailing whitespace")]
111 SurroundingWhitespace,
112 #[error("datasource name must use ASCII letters, digits, '.', '_' or '-' and stay within the byte limit")]
114 InvalidCharacters,
115}
116
117#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
119pub struct DatasourceRef(Arc<str>);
120
121impl DatasourceRef {
122 pub fn new(name: impl AsRef<str>) -> anyhow::Result<Self> {
129 Self::try_new(name).map_err(anyhow::Error::new)
130 }
131
132 pub fn try_new(name: impl AsRef<str>) -> Result<Self, DatasourceNameError> {
139 let name = name.as_ref();
140 validate_datasource_name(name)?;
141 Ok(Self(Arc::from(name)))
142 }
143
144 pub fn default_ref() -> Self {
150 Self(Arc::from(DEFAULT_DATASOURCE))
151 }
152
153 pub fn as_str(&self) -> &str {
159 &self.0
160 }
161}
162
163impl Default for DatasourceRef {
164 fn default() -> Self {
170 Self::default_ref()
171 }
172}
173
174impl AsRef<str> for DatasourceRef {
175 fn as_ref(&self) -> &str {
181 self.as_str()
182 }
183}
184
185impl std::fmt::Display for DatasourceRef {
186 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
193 formatter.write_str(self.as_str())
194 }
195}
196
197pub fn validate_datasource_name(name: &str) -> Result<(), DatasourceNameError> {
204 if name.is_empty() {
205 return Err(DatasourceNameError::Empty);
206 }
207 if name.trim() != name {
208 return Err(DatasourceNameError::SurroundingWhitespace);
209 }
210 if name.len() > MAX_DATASOURCE_NAME_BYTES
211 || !name
212 .bytes()
213 .all(|value| value.is_ascii_alphanumeric() || matches!(value, b'.' | b'_' | b'-'))
214 {
215 return Err(DatasourceNameError::InvalidCharacters);
216 }
217 Ok(())
218}
219
220#[derive(Clone, serde::Deserialize)]
222#[serde(deny_unknown_fields)]
223pub struct DataSourcePoolConfig {
224 pub url: String,
226 #[serde(default = "default_max_connections")]
228 pub max_connections: u32,
229 #[serde(default)]
231 pub min_connections: u32,
232 #[serde(default = "default_acquire_timeout_ms")]
234 pub acquire_timeout_ms: u64,
235 #[serde(default = "default_connect_timeout_ms")]
237 pub connect_timeout_ms: u64,
238 #[serde(default = "default_probe_on_start")]
240 pub probe_on_start: bool,
241}
242
243impl std::fmt::Debug for DataSourcePoolConfig {
244 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
251 formatter
252 .debug_struct("DataSourcePoolConfig")
253 .field("url", &redact_url(&self.url))
254 .field("max_connections", &self.max_connections)
255 .field("min_connections", &self.min_connections)
256 .field("acquire_timeout_ms", &self.acquire_timeout_ms)
257 .field("connect_timeout_ms", &self.connect_timeout_ms)
258 .field("probe_on_start", &self.probe_on_start)
259 .finish()
260 }
261}
262
263#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
265pub enum PoolConfigError {
266 #[error("datasource url cannot be empty")]
268 EmptyUrl,
269 #[error("datasource max_connections must be greater than zero")]
271 ZeroMaxConnections,
272 #[error("datasource min_connections cannot exceed max_connections")]
274 InvalidConnectionRange,
275 #[error("datasource acquire_timeout_ms must be greater than zero")]
277 ZeroAcquireTimeout,
278 #[error("datasource connect_timeout_ms must be greater than zero")]
280 ZeroConnectTimeout,
281}
282
283impl DataSourcePoolConfig {
284 pub fn validate_common(&self) -> Result<(), PoolConfigError> {
290 if self.url.trim().is_empty() {
291 return Err(PoolConfigError::EmptyUrl);
292 }
293 if self.max_connections == 0 {
294 return Err(PoolConfigError::ZeroMaxConnections);
295 }
296 if self.min_connections > self.max_connections {
297 return Err(PoolConfigError::InvalidConnectionRange);
298 }
299 if self.acquire_timeout_ms == 0 {
300 return Err(PoolConfigError::ZeroAcquireTimeout);
301 }
302 if self.connect_timeout_ms == 0 {
303 return Err(PoolConfigError::ZeroConnectTimeout);
304 }
305 Ok(())
306 }
307
308 pub fn endpoint(&self) -> String {
314 redacted_endpoint(&self.url)
315 }
316}
317
318pub fn redact_url(url: &str) -> String {
325 let Some((scheme, rest)) = url.split_once("://") else {
326 return "<redacted-url>".to_owned();
327 };
328 if scheme.is_empty()
329 || !scheme
330 .bytes()
331 .all(|value| value.is_ascii_alphanumeric() || matches!(value, b'+' | b'-' | b'.'))
332 {
333 return "<redacted-url>".to_owned();
334 }
335 let endpoint = redacted_endpoint(url);
336 if endpoint == "<unknown-endpoint>" {
337 return "<redacted-url>".to_owned();
338 }
339 let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
340 if authority.contains('@') {
341 format!("{scheme}://***@{endpoint}")
342 } else {
343 format!("{scheme}://{endpoint}")
344 }
345}
346
347pub fn redacted_endpoint(url: &str) -> String {
354 let Some((_, rest)) = url.split_once("://") else {
355 return "<unknown-endpoint>".to_owned();
356 };
357 let authority_and_path = rest.rsplit_once('@').map_or(rest, |(_, value)| value);
358 let endpoint = authority_and_path
359 .split(['?', '#'])
360 .next()
361 .unwrap_or(authority_and_path);
362 if endpoint.is_empty() {
363 "<unknown-endpoint>".to_owned()
364 } else {
365 endpoint.to_owned()
366 }
367}
368
369fn default_max_connections() -> u32 {
375 10
376}
377
378fn default_acquire_timeout_ms() -> u64 {
384 2_000
385}
386
387fn default_connect_timeout_ms() -> u64 {
393 5_000
394}
395
396fn default_probe_on_start() -> bool {
402 true
403}
404
405static NEXT_OWNER_ID: AtomicU64 = AtomicU64::new(1);
406
407#[derive(Debug)]
408struct OwnerIdentity {
410 id: u64,
411}
412
413#[derive(Clone)]
415pub struct ManagedRegistryOwner(Arc<OwnerIdentity>);
416
417impl ManagedRegistryOwner {
418 pub fn new() -> Self {
424 Self(Arc::new(OwnerIdentity {
425 id: NEXT_OWNER_ID.fetch_add(1, Ordering::Relaxed),
426 }))
427 }
428
429 pub fn ptr_eq(&self, other: &Self) -> bool {
436 Arc::ptr_eq(&self.0, &other.0)
437 }
438
439 pub fn diagnostic_id(&self) -> u64 {
445 self.0.id
446 }
447
448 fn downgrade(&self) -> Weak<OwnerIdentity> {
454 Arc::downgrade(&self.0)
455 }
456}
457
458impl Default for ManagedRegistryOwner {
459 fn default() -> Self {
465 Self::new()
466 }
467}
468
469impl std::fmt::Debug for ManagedRegistryOwner {
470 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
477 formatter
478 .debug_tuple("ManagedRegistryOwner")
479 .field(&self.diagnostic_id())
480 .finish()
481 }
482}
483
484#[derive(Debug, Clone, Copy, PartialEq, Eq)]
486pub enum BootstrapKind {
487 Configured,
489 DeferredDefault,
491}
492
493#[derive(Clone, Debug)]
495pub struct ManagedInstallationToken {
496 owner: ManagedRegistryOwner,
497 kind: BootstrapKind,
498}
499
500impl ManagedInstallationToken {
501 pub fn owner(&self) -> &ManagedRegistryOwner {
507 &self.owner
508 }
509
510 pub fn kind(&self) -> BootstrapKind {
516 self.kind
517 }
518}
519
520pub struct DataSourceCatalog {
522 owner: ManagedRegistryOwner,
523 entries: BTreeMap<DatasourceRef, DatabaseDriver>,
524 lifecycle: AtomicU8,
525}
526
527const CATALOG_STAGING: u8 = 0;
528const CATALOG_OPEN: u8 = 1;
529const CATALOG_STOPPED: u8 = 2;
530
531impl DataSourceCatalog {
532 pub fn try_new(
540 owner: ManagedRegistryOwner,
541 entries: impl IntoIterator<Item = (String, DatabaseDriver)>,
542 ) -> Result<Self, CatalogBuildError> {
543 let mut catalog = BTreeMap::new();
544 for (name, driver) in entries {
545 if catalog.len() >= MAX_MANAGED_DATASOURCES {
546 return Err(CatalogBuildError::TooManyDatasources);
547 }
548 let reference = DatasourceRef::try_new(name).map_err(CatalogBuildError::InvalidName)?;
549 if catalog.insert(reference.clone(), driver).is_some() {
550 return Err(CatalogBuildError::Duplicate(reference));
551 }
552 }
553 if catalog.is_empty() {
554 return Err(CatalogBuildError::Empty);
555 }
556 Ok(Self {
557 owner,
558 entries: catalog,
559 lifecycle: AtomicU8::new(CATALOG_STAGING),
560 })
561 }
562
563 pub fn owner(&self) -> &ManagedRegistryOwner {
569 &self.owner
570 }
571
572 pub fn len(&self) -> usize {
578 self.entries.len()
579 }
580
581 pub fn is_empty(&self) -> bool {
587 self.entries.is_empty()
588 }
589
590 pub fn entries(&self) -> Vec<(DatasourceRef, DatabaseDriver)> {
596 self.entries
597 .iter()
598 .map(|(reference, driver)| (reference.clone(), *driver))
599 .collect()
600 }
601
602 pub fn resolve(
610 &self,
611 datasource: &str,
612 expected: DatabaseDriver,
613 ) -> Result<DatasourceRef, DataSourceLookupError> {
614 match self.lifecycle.load(Ordering::Acquire) {
615 CATALOG_OPEN => {}
616 CATALOG_STOPPED => return Err(DataSourceLookupError::Stopped),
617 _ => return Err(DataSourceLookupError::RegistryUnavailable),
618 }
619 let reference =
620 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)?;
621 let actual = self.entries.get(&reference).copied().ok_or_else(|| {
622 DataSourceLookupError::NotFound {
623 datasource: reference.clone(),
624 }
625 })?;
626 if actual != expected {
627 return Err(DataSourceLookupError::DriverMismatch {
628 datasource: reference,
629 expected,
630 actual,
631 });
632 }
633 Ok(reference)
634 }
635
636 fn open(&self) {
642 self.lifecycle.store(CATALOG_OPEN, Ordering::Release);
643 }
644
645 fn stop_accepting(&self) {
651 self.lifecycle.store(CATALOG_STOPPED, Ordering::Release);
652 }
653}
654
655#[derive(Debug, thiserror::Error)]
657pub enum CatalogBuildError {
658 #[error("managed datasource catalog cannot be empty")]
660 Empty,
661 #[error("managed datasource count exceeds the supported limit")]
663 TooManyDatasources,
664 #[error("invalid datasource name")]
666 InvalidName(#[source] DatasourceNameError),
667 #[error("datasource `{0}` is configured more than once")]
669 Duplicate(DatasourceRef),
670}
671
672#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
674pub enum DataSourceLookupError {
675 #[error("invalid datasource name")]
677 InvalidName,
678 #[error("datasource registry is unavailable")]
680 RegistryUnavailable,
681 #[error("datasource registry is stopped")]
683 Stopped,
684 #[error("datasource `{datasource}` is not managed by this application")]
686 NotFound {
687 datasource: DatasourceRef,
689 },
690 #[error("datasource `{datasource}` uses {actual}, not {expected}")]
692 DriverMismatch {
693 datasource: DatasourceRef,
695 expected: DatabaseDriver,
697 actual: DatabaseDriver,
699 },
700}
701
702#[must_use]
704#[derive(Debug, Clone, PartialEq, Eq)]
705pub enum TxDecision<T, E> {
706 Commit(T),
708 Rollback(E),
710}
711
712#[derive(Debug)]
714pub enum TxRollbackCause<E> {
715 Decision(E),
717 RollbackOnly {
719 reason: String,
721 },
722}
723
724#[derive(Debug)]
726pub enum TxRunError<E> {
727 Rollback(E),
729 RollbackOnly {
731 reason: String,
733 },
734 CommitRejected {
736 reason: String,
738 },
739 CommitUncertain {
741 reason: String,
743 },
744 RollbackFailed {
746 cause: TxRollbackCause<E>,
748 reason: String,
750 },
751 Infrastructure {
753 reason: String,
755 },
756}
757
758#[derive(Debug)]
760pub enum TxEntryError<E> {
761 Lookup(DataSourceLookupError),
763 Run(TxRunError<E>),
765}
766
767impl<E: std::fmt::Display> std::fmt::Display for TxEntryError<E> {
768 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
775 match self {
776 Self::Lookup(error) => {
777 write!(formatter, "transaction datasource lookup failed: {error}")
778 }
779 Self::Run(error) => std::fmt::Display::fmt(error, formatter),
780 }
781 }
782}
783
784impl<E> std::error::Error for TxEntryError<E> where E: std::fmt::Debug + std::fmt::Display {}
785
786#[derive(Debug, Clone, Copy, PartialEq, Eq)]
788pub enum CommitOutcome {
789 Committed,
791 Rejected,
793 OutcomeUnknown,
795}
796
797impl<E: std::fmt::Display> std::fmt::Display for TxRunError<E> {
798 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
805 match self {
806 Self::Rollback(error) => write!(formatter, "transaction rolled back: {error}"),
807 Self::RollbackOnly {
808 reason,
809 } => {
810 write!(formatter, "transaction was rollback-only: {reason}")
811 }
812 Self::CommitRejected {
813 reason,
814 } => write!(formatter, "commit rejected: {reason}"),
815 Self::CommitUncertain {
816 reason,
817 } => write!(formatter, "commit uncertain: {reason}"),
818 Self::RollbackFailed {
819 reason,
820 ..
821 } => write!(formatter, "rollback failed: {reason}"),
822 Self::Infrastructure {
823 reason,
824 } => {
825 write!(formatter, "transaction infrastructure failed: {reason}")
826 }
827 }
828 }
829}
830
831impl<E> std::error::Error for TxRunError<E> where E: std::fmt::Debug + std::fmt::Display {}
832
833#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
835#[error("ambient transaction driver mismatch: active={active}, requested={requested}")]
836pub struct DriverScopeError {
837 pub active: DatabaseDriver,
839 pub requested: DatabaseDriver,
841}
842
843tokio::task_local! {
844 static ACTIVE_DRIVER: DatabaseDriver;
845}
846
847pub fn current_driver() -> Option<DatabaseDriver> {
853 ACTIVE_DRIVER.try_with(|driver| *driver).ok()
854}
855
856pub fn ensure_driver(requested: DatabaseDriver) -> Result<(), DriverScopeError> {
863 match current_driver() {
864 Some(active) if active != requested => Err(DriverScopeError {
865 active,
866 requested,
867 }),
868 _ => Ok(()),
869 }
870}
871
872pub async fn scope_driver<F>(
880 driver: DatabaseDriver,
881 future: F,
882) -> Result<F::Output, DriverScopeError>
883where
884 F: Future,
885{
886 match current_driver() {
887 Some(active) if active != driver => Err(DriverScopeError {
888 active,
889 requested: driver,
890 }),
891 Some(_) => Ok(future.await),
892 None => Ok(ACTIVE_DRIVER.scope(driver, future).await),
893 }
894}
895
896#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
898pub enum RegistryCoordinationError {
899 #[error("standalone datasource driver conflicts with the active driver")]
901 StandaloneDriverConflict,
902 #[error("standalone datasource registration conflicts with managed application state")]
904 ManagedModeConflict,
905 #[error("deferred datasource bootstrap accepts only the default datasource")]
907 DeferredOnlyDefault,
908 #[error("deferred datasource bootstrap already uses another driver")]
910 DeferredDriverConflict,
911 #[error("datasource registry is already occupied")]
913 RegistryOccupied,
914 #[error("managed registry owner token does not match the active bootstrap")]
916 OwnerMismatch,
917 #[error("datasource catalog owner does not match the installation token")]
919 CatalogOwnerMismatch,
920 #[error("datasource catalog identity does not match the active managed catalog")]
922 CatalogIdentityMismatch,
923 #[error("deferred datasource bootstrap did not install the expected default driver")]
925 DeferredDriverMissing,
926 #[error("deferred datasource resources must be removed before bootstrap cleanup")]
928 DeferredResourcesRemain,
929 #[error("public standalone adopt is unavailable during application bootstrap")]
931 PublicAdoptDuringBootstrap,
932 #[error("orphaned deferred datasource resources require explicit cleanup")]
934 OrphanedDeferredResources,
935 #[error("deferred datasource catalog must contain only the installed default driver")]
937 InvalidDeferredCatalog,
938}
939
940#[derive(Debug)]
941enum RegistryMode {
942 Empty,
943 Standalone(DatabaseDriver),
944 Bootstrapping {
945 owner: Weak<OwnerIdentity>,
946 kind: BootstrapKind,
947 deferred_driver: Option<DatabaseDriver>,
948 },
949 Managed {
950 owner: Weak<OwnerIdentity>,
951 catalog: Weak<DataSourceCatalog>,
952 },
953}
954
955#[derive(Debug)]
956struct RegistryState {
958 mode: RegistryMode,
959}
960
961impl Default for RegistryState {
962 fn default() -> Self {
968 Self {
969 mode: RegistryMode::Empty,
970 }
971 }
972}
973
974static REGISTRY_STATE: OnceLock<Mutex<RegistryState>> = OnceLock::new();
975
976#[derive(Debug, Clone, Copy, PartialEq, Eq)]
978pub enum RegistryModeSnapshot {
979 Empty,
981 Standalone(DatabaseDriver),
983 Bootstrapping {
985 kind: BootstrapKind,
987 deferred_driver: Option<DatabaseDriver>,
989 },
990 Managed,
992 OrphanedDeferred(DatabaseDriver),
994}
995
996#[doc(hidden)]
1003pub fn coordinate<T>(operation: impl FnOnce(&mut RegistryCoordinator<'_>) -> T) -> T {
1004 let mut state = REGISTRY_STATE
1005 .get_or_init(|| Mutex::new(RegistryState::default()))
1006 .lock()
1007 .unwrap_or_else(|error| error.into_inner());
1008 let mut coordinator = RegistryCoordinator {
1009 state: &mut state,
1010 };
1011 coordinator.sanitize();
1012 operation(&mut coordinator)
1013}
1014
1015#[doc(hidden)]
1017pub struct RegistryCoordinator<'a> {
1018 state: &'a mut RegistryState,
1019}
1020
1021impl RegistryCoordinator<'_> {
1022 fn sanitize(&mut self) {
1028 match &self.state.mode {
1029 RegistryMode::Bootstrapping {
1030 owner,
1031 deferred_driver: None,
1032 ..
1033 } if owner.upgrade().is_none() => {
1034 self.state.mode = RegistryMode::Empty;
1035 }
1036 RegistryMode::Managed {
1037 owner,
1038 catalog,
1039 } if owner.upgrade().is_none() && catalog.upgrade().is_none() => {
1040 self.state.mode = RegistryMode::Empty;
1041 }
1042 _ => {}
1043 }
1044 }
1045
1046 pub fn register_standalone(
1054 &mut self,
1055 driver: DatabaseDriver,
1056 datasource: &str,
1057 ) -> Result<(), RegistryCoordinationError> {
1058 match &mut self.state.mode {
1059 RegistryMode::Empty => {
1060 self.state.mode = RegistryMode::Standalone(driver);
1061 Ok(())
1062 }
1063 RegistryMode::Standalone(active) if *active == driver => Ok(()),
1064 RegistryMode::Standalone(_) => Err(RegistryCoordinationError::StandaloneDriverConflict),
1065 RegistryMode::Bootstrapping {
1066 owner,
1067 kind: BootstrapKind::DeferredDefault,
1068 deferred_driver,
1069 } => {
1070 if owner.upgrade().is_none() && deferred_driver.is_some() {
1071 return Err(RegistryCoordinationError::OrphanedDeferredResources);
1072 }
1073 if datasource != DEFAULT_DATASOURCE {
1074 return Err(RegistryCoordinationError::DeferredOnlyDefault);
1075 }
1076 match deferred_driver {
1077 Some(active) if *active != driver => {
1078 Err(RegistryCoordinationError::DeferredDriverConflict)
1079 }
1080 Some(_) => Ok(()),
1081 slot @ None => {
1082 *slot = Some(driver);
1083 Ok(())
1084 }
1085 }
1086 }
1087 RegistryMode::Bootstrapping {
1088 ..
1089 }
1090 | RegistryMode::Managed {
1091 ..
1092 } => Err(RegistryCoordinationError::ManagedModeConflict),
1093 }
1094 }
1095
1096 pub fn release_standalone(
1103 &mut self,
1104 driver: DatabaseDriver,
1105 ) -> Result<(), RegistryCoordinationError> {
1106 match &mut self.state.mode {
1107 RegistryMode::Standalone(active) if *active == driver => {
1108 self.state.mode = RegistryMode::Empty;
1109 Ok(())
1110 }
1111 RegistryMode::Bootstrapping {
1112 owner,
1113 kind: BootstrapKind::DeferredDefault,
1114 deferred_driver: Some(active),
1115 } if *active == driver => {
1116 let owner_alive = owner.upgrade().is_some();
1117 if owner_alive {
1118 if let RegistryMode::Bootstrapping {
1119 deferred_driver,
1120 ..
1121 } = &mut self.state.mode
1122 {
1123 *deferred_driver = None;
1124 }
1125 } else {
1126 self.state.mode = RegistryMode::Empty;
1127 }
1128 Ok(())
1129 }
1130 RegistryMode::Empty => Ok(()),
1131 _ => Err(RegistryCoordinationError::ManagedModeConflict),
1132 }
1133 }
1134
1135 pub fn begin_bootstrap(
1143 &mut self,
1144 owner: &ManagedRegistryOwner,
1145 kind: BootstrapKind,
1146 ) -> Result<ManagedInstallationToken, RegistryCoordinationError> {
1147 if !matches!(self.state.mode, RegistryMode::Empty) {
1148 return Err(RegistryCoordinationError::RegistryOccupied);
1149 }
1150 self.state.mode = RegistryMode::Bootstrapping {
1151 owner: owner.downgrade(),
1152 kind,
1153 deferred_driver: None,
1154 };
1155 Ok(ManagedInstallationToken {
1156 owner: owner.clone(),
1157 kind,
1158 })
1159 }
1160
1161 pub fn begin_legacy_adopt(
1169 &mut self,
1170 owner: &ManagedRegistryOwner,
1171 driver: DatabaseDriver,
1172 ) -> Result<ManagedInstallationToken, RegistryCoordinationError> {
1173 match &self.state.mode {
1174 RegistryMode::Standalone(active) if *active == driver => {
1175 self.state.mode = RegistryMode::Bootstrapping {
1176 owner: owner.downgrade(),
1177 kind: BootstrapKind::Configured,
1178 deferred_driver: None,
1179 };
1180 Ok(ManagedInstallationToken {
1181 owner: owner.clone(),
1182 kind: BootstrapKind::Configured,
1183 })
1184 }
1185 RegistryMode::Bootstrapping {
1186 ..
1187 } => Err(RegistryCoordinationError::PublicAdoptDuringBootstrap),
1188 _ => Err(RegistryCoordinationError::RegistryOccupied),
1189 }
1190 }
1191
1192 pub fn restore_legacy_standalone(
1200 &mut self,
1201 token: &ManagedInstallationToken,
1202 driver: DatabaseDriver,
1203 ) -> Result<(), RegistryCoordinationError> {
1204 self.verify_bootstrap_owner(token)?;
1205 self.state.mode = RegistryMode::Standalone(driver);
1206 Ok(())
1207 }
1208
1209 pub fn verify_deferred_adopt(
1217 &mut self,
1218 token: &ManagedInstallationToken,
1219 driver: DatabaseDriver,
1220 ) -> Result<(), RegistryCoordinationError> {
1221 self.verify_bootstrap_owner(token)?;
1222 match &self.state.mode {
1223 RegistryMode::Bootstrapping {
1224 kind: BootstrapKind::DeferredDefault,
1225 deferred_driver: Some(active),
1226 ..
1227 } if *active == driver => Ok(()),
1228 _ => Err(RegistryCoordinationError::DeferredDriverMissing),
1229 }
1230 }
1231
1232 pub fn verify_installation_token(
1239 &self,
1240 token: &ManagedInstallationToken,
1241 ) -> Result<(), RegistryCoordinationError> {
1242 self.verify_bootstrap_owner(token)
1243 }
1244
1245 pub fn install_catalog(
1253 &mut self,
1254 token: &ManagedInstallationToken,
1255 catalog: &Arc<DataSourceCatalog>,
1256 ) -> Result<(), RegistryCoordinationError> {
1257 self.verify_bootstrap_owner(token)?;
1258 if !catalog.owner().ptr_eq(token.owner()) {
1259 return Err(RegistryCoordinationError::CatalogOwnerMismatch);
1260 }
1261 if let RegistryMode::Bootstrapping {
1262 kind: BootstrapKind::DeferredDefault,
1263 deferred_driver: Some(driver),
1264 ..
1265 } = &self.state.mode
1266 {
1267 let entries = catalog.entries();
1268 if entries.len() != 1
1269 || entries[0].0.as_str() != DEFAULT_DATASOURCE
1270 || entries[0].1 != *driver
1271 {
1272 return Err(RegistryCoordinationError::InvalidDeferredCatalog);
1273 }
1274 }
1275 self.state.mode = RegistryMode::Managed {
1276 owner: token.owner.downgrade(),
1277 catalog: Arc::downgrade(catalog),
1278 };
1279 Ok(())
1280 }
1281
1282 pub fn open_catalog(
1290 &mut self,
1291 token: &ManagedInstallationToken,
1292 catalog: &Arc<DataSourceCatalog>,
1293 ) -> Result<(), RegistryCoordinationError> {
1294 match &self.state.mode {
1295 RegistryMode::Managed {
1296 owner,
1297 catalog: active,
1298 } => {
1299 let Some(active_owner) = owner.upgrade() else {
1300 return Err(RegistryCoordinationError::OwnerMismatch);
1301 };
1302 if !Arc::ptr_eq(&active_owner, &token.owner.0) {
1303 return Err(RegistryCoordinationError::OwnerMismatch);
1304 }
1305 let Some(active_catalog) = active.upgrade() else {
1306 return Err(RegistryCoordinationError::CatalogIdentityMismatch);
1307 };
1308 if !Arc::ptr_eq(&active_catalog, catalog) {
1309 return Err(RegistryCoordinationError::CatalogIdentityMismatch);
1310 }
1311 catalog.open();
1312 Ok(())
1313 }
1314 _ => Err(RegistryCoordinationError::OwnerMismatch),
1315 }
1316 }
1317
1318 pub fn clear_managed(
1325 &mut self,
1326 owner: &ManagedRegistryOwner,
1327 ) -> Result<(), RegistryCoordinationError> {
1328 match &self.state.mode {
1329 RegistryMode::Managed {
1330 owner: active,
1331 catalog,
1332 } => {
1333 let Some(active_owner) = active.upgrade() else {
1334 return Err(RegistryCoordinationError::OwnerMismatch);
1335 };
1336 if !Arc::ptr_eq(&active_owner, &owner.0) {
1337 return Err(RegistryCoordinationError::OwnerMismatch);
1338 }
1339 if let Some(catalog) = catalog.upgrade() {
1340 catalog.stop_accepting();
1341 }
1342 self.state.mode = RegistryMode::Empty;
1343 Ok(())
1344 }
1345 _ => Err(RegistryCoordinationError::OwnerMismatch),
1346 }
1347 }
1348
1349 pub fn clear_deferred_driver(
1357 &mut self,
1358 token: &ManagedInstallationToken,
1359 driver: DatabaseDriver,
1360 ) -> Result<(), RegistryCoordinationError> {
1361 self.verify_bootstrap_owner(token)?;
1362 match &mut self.state.mode {
1363 RegistryMode::Bootstrapping {
1364 kind: BootstrapKind::DeferredDefault,
1365 deferred_driver: Some(active),
1366 ..
1367 } if *active == driver => {
1368 if let RegistryMode::Bootstrapping {
1369 deferred_driver,
1370 ..
1371 } = &mut self.state.mode
1372 {
1373 *deferred_driver = None;
1374 }
1375 Ok(())
1376 }
1377 _ => Err(RegistryCoordinationError::DeferredDriverMissing),
1378 }
1379 }
1380
1381 pub fn abort_bootstrap(
1388 &mut self,
1389 token: &ManagedInstallationToken,
1390 ) -> Result<(), RegistryCoordinationError> {
1391 self.verify_bootstrap_owner(token)?;
1392 match &self.state.mode {
1393 RegistryMode::Bootstrapping {
1394 deferred_driver: Some(_),
1395 ..
1396 } => Err(RegistryCoordinationError::DeferredResourcesRemain),
1397 RegistryMode::Bootstrapping {
1398 ..
1399 } => {
1400 self.state.mode = RegistryMode::Empty;
1401 Ok(())
1402 }
1403 _ => Err(RegistryCoordinationError::OwnerMismatch),
1404 }
1405 }
1406
1407 pub fn snapshot(&self) -> RegistryModeSnapshot {
1413 match &self.state.mode {
1414 RegistryMode::Empty => RegistryModeSnapshot::Empty,
1415 RegistryMode::Standalone(driver) => RegistryModeSnapshot::Standalone(*driver),
1416 RegistryMode::Bootstrapping {
1417 owner,
1418 kind,
1419 deferred_driver,
1420 } => match (owner.upgrade(), deferred_driver) {
1421 (None, Some(driver)) => RegistryModeSnapshot::OrphanedDeferred(*driver),
1422 _ => RegistryModeSnapshot::Bootstrapping {
1423 kind: *kind,
1424 deferred_driver: *deferred_driver,
1425 },
1426 },
1427 RegistryMode::Managed {
1428 ..
1429 } => RegistryModeSnapshot::Managed,
1430 }
1431 }
1432
1433 fn verify_bootstrap_owner(
1440 &self,
1441 token: &ManagedInstallationToken,
1442 ) -> Result<(), RegistryCoordinationError> {
1443 match &self.state.mode {
1444 RegistryMode::Bootstrapping {
1445 owner,
1446 kind,
1447 ..
1448 } => {
1449 let Some(active) = owner.upgrade() else {
1450 return Err(RegistryCoordinationError::OwnerMismatch);
1451 };
1452 if Arc::ptr_eq(&active, &token.owner.0) && *kind == token.kind {
1453 Ok(())
1454 } else {
1455 Err(RegistryCoordinationError::OwnerMismatch)
1456 }
1457 }
1458 _ => Err(RegistryCoordinationError::OwnerMismatch),
1459 }
1460 }
1461}
1462
1463pub fn registry_mode() -> RegistryModeSnapshot {
1469 coordinate(|coordinator| coordinator.snapshot())
1470}
1471
1472pub fn begin_managed_bootstrap(
1479 kind: BootstrapKind,
1480) -> Result<(ManagedRegistryOwner, ManagedInstallationToken), RegistryCoordinationError> {
1481 let owner = ManagedRegistryOwner::new();
1482 let token = coordinate(|coordinator| coordinator.begin_bootstrap(&owner, kind))?;
1483 Ok((owner, token))
1484}
1485
1486pub fn install_managed_catalog(
1494 token: &ManagedInstallationToken,
1495 catalog: &Arc<DataSourceCatalog>,
1496) -> Result<(), RegistryCoordinationError> {
1497 coordinate(|coordinator| coordinator.install_catalog(token, catalog))
1498}
1499
1500pub fn open_managed_catalog(
1508 token: &ManagedInstallationToken,
1509 catalog: &Arc<DataSourceCatalog>,
1510) -> Result<(), RegistryCoordinationError> {
1511 coordinate(|coordinator| coordinator.open_catalog(token, catalog))
1512}
1513
1514pub fn clear_managed_catalog(
1521 owner: &ManagedRegistryOwner,
1522) -> Result<(), RegistryCoordinationError> {
1523 coordinate(|coordinator| coordinator.clear_managed(owner))
1524}
1525
1526pub fn abort_managed_bootstrap(
1533 token: &ManagedInstallationToken,
1534) -> Result<(), RegistryCoordinationError> {
1535 coordinate(|coordinator| coordinator.abort_bootstrap(token))
1536}
1537
1538pub fn resolve_datasource(
1547 datasource: &str,
1548 expected: DatabaseDriver,
1549) -> Result<DatasourceRef, DataSourceLookupError> {
1550 coordinate(|coordinator| match &coordinator.state.mode {
1551 RegistryMode::Empty => {
1552 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
1555 }
1556 RegistryMode::Bootstrapping {
1557 kind: BootstrapKind::DeferredDefault,
1558 deferred_driver: Some(actual),
1559 ..
1560 } if *actual == expected && datasource == DEFAULT_DATASOURCE => {
1561 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
1564 }
1565 RegistryMode::Bootstrapping {
1566 ..
1567 } => Err(DataSourceLookupError::RegistryUnavailable),
1568 RegistryMode::Standalone(actual) if *actual == expected => {
1569 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
1570 }
1571 RegistryMode::Standalone(actual) => {
1572 let reference =
1573 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)?;
1574 Err(DataSourceLookupError::DriverMismatch {
1575 datasource: reference,
1576 expected,
1577 actual: *actual,
1578 })
1579 }
1580 RegistryMode::Managed {
1581 catalog,
1582 ..
1583 } => catalog
1584 .upgrade()
1585 .ok_or(DataSourceLookupError::RegistryUnavailable)?
1586 .resolve(datasource, expected),
1587 })
1588}
1589
1590pub fn managed_catalog() -> Result<Arc<DataSourceCatalog>, DataSourceLookupError> {
1596 coordinate(|coordinator| match &coordinator.state.mode {
1597 RegistryMode::Managed {
1598 catalog,
1599 ..
1600 } => catalog
1601 .upgrade()
1602 .ok_or(DataSourceLookupError::RegistryUnavailable)
1603 .and_then(|catalog| match catalog.lifecycle.load(Ordering::Acquire) {
1604 CATALOG_OPEN => Ok(catalog),
1605 CATALOG_STOPPED => Err(DataSourceLookupError::Stopped),
1606 _ => Err(DataSourceLookupError::RegistryUnavailable),
1607 }),
1608 _ => Err(DataSourceLookupError::RegistryUnavailable),
1609 })
1610}