1use std::collections::BTreeMap;
9use std::future::Future;
10use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
11use std::sync::{Arc, Mutex, OnceLock, Weak};
12
13pub const DEFAULT_DATASOURCE: &str = "default";
15
16pub mod observability;
17pub const MAX_DATASOURCE_NAME_BYTES: usize = 128;
19pub const MAX_MANAGED_DATASOURCES: usize = 64;
21
22#[derive(
24 Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Deserialize, serde::Serialize,
25)]
26#[serde(rename_all = "lowercase")]
27pub enum DatabaseDriver {
28 MySql,
30 PostgreSql,
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
36pub struct DatabaseCapabilities(u8);
37
38impl DatabaseCapabilities {
39 const MYSQL: u8 = 1;
40 const POSTGRESQL: u8 = 1 << 1;
41
42 pub const fn empty() -> Self {
48 Self(0)
49 }
50
51 pub const fn with(self, driver: DatabaseDriver) -> Self {
57 let flag = match driver {
58 DatabaseDriver::MySql => Self::MYSQL,
59 DatabaseDriver::PostgreSql => Self::POSTGRESQL,
60 };
61 Self(self.0 | flag)
62 }
63
64 pub const fn contains(self, driver: DatabaseDriver) -> bool {
70 let flag = match driver {
71 DatabaseDriver::MySql => Self::MYSQL,
72 DatabaseDriver::PostgreSql => Self::POSTGRESQL,
73 };
74 self.0 & flag != 0
75 }
76}
77
78impl std::fmt::Display for DatabaseDriver {
79 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85 formatter.write_str(match self {
86 Self::MySql => "mysql",
87 Self::PostgreSql => "postgresql",
88 })
89 }
90}
91
92#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
94pub enum DatasourceNameError {
95 #[error("datasource name cannot be empty")]
97 Empty,
98 #[error("datasource name cannot contain leading or trailing whitespace")]
100 SurroundingWhitespace,
101 #[error("datasource name must use ASCII letters, digits, '.', '_' or '-' and stay within the byte limit")]
103 InvalidCharacters,
104}
105
106#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
108pub struct DatasourceRef(Arc<str>);
109
110impl DatasourceRef {
111 pub fn new(name: impl AsRef<str>) -> anyhow::Result<Self> {
117 Self::try_new(name).map_err(anyhow::Error::new)
118 }
119
120 pub fn try_new(name: impl AsRef<str>) -> Result<Self, DatasourceNameError> {
126 let name = name.as_ref();
127 validate_datasource_name(name)?;
128 Ok(Self(Arc::from(name)))
129 }
130
131 pub fn default_ref() -> Self {
137 Self(Arc::from(DEFAULT_DATASOURCE))
138 }
139
140 pub fn as_str(&self) -> &str {
146 &self.0
147 }
148}
149
150impl Default for DatasourceRef {
151 fn default() -> Self {
157 Self::default_ref()
158 }
159}
160
161impl AsRef<str> for DatasourceRef {
162 fn as_ref(&self) -> &str {
168 self.as_str()
169 }
170}
171
172impl std::fmt::Display for DatasourceRef {
173 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
179 formatter.write_str(self.as_str())
180 }
181}
182
183pub fn validate_datasource_name(name: &str) -> Result<(), DatasourceNameError> {
189 if name.is_empty() {
190 return Err(DatasourceNameError::Empty);
191 }
192 if name.trim() != name {
193 return Err(DatasourceNameError::SurroundingWhitespace);
194 }
195 if name.len() > MAX_DATASOURCE_NAME_BYTES
196 || !name
197 .bytes()
198 .all(|value| value.is_ascii_alphanumeric() || matches!(value, b'.' | b'_' | b'-'))
199 {
200 return Err(DatasourceNameError::InvalidCharacters);
201 }
202 Ok(())
203}
204
205#[derive(Clone, serde::Deserialize)]
207#[serde(deny_unknown_fields)]
208pub struct DataSourcePoolConfig {
209 pub url: String,
211 #[serde(default = "default_max_connections")]
213 pub max_connections: u32,
214 #[serde(default)]
216 pub min_connections: u32,
217 #[serde(default = "default_acquire_timeout_ms")]
219 pub acquire_timeout_ms: u64,
220 #[serde(default = "default_connect_timeout_ms")]
222 pub connect_timeout_ms: u64,
223 #[serde(default = "default_probe_on_start")]
225 pub probe_on_start: bool,
226}
227
228impl std::fmt::Debug for DataSourcePoolConfig {
229 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
235 formatter
236 .debug_struct("DataSourcePoolConfig")
237 .field("url", &redact_url(&self.url))
238 .field("max_connections", &self.max_connections)
239 .field("min_connections", &self.min_connections)
240 .field("acquire_timeout_ms", &self.acquire_timeout_ms)
241 .field("connect_timeout_ms", &self.connect_timeout_ms)
242 .field("probe_on_start", &self.probe_on_start)
243 .finish()
244 }
245}
246
247#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
249pub enum PoolConfigError {
250 #[error("datasource url cannot be empty")]
252 EmptyUrl,
253 #[error("datasource max_connections must be greater than zero")]
255 ZeroMaxConnections,
256 #[error("datasource min_connections cannot exceed max_connections")]
258 InvalidConnectionRange,
259 #[error("datasource acquire_timeout_ms must be greater than zero")]
261 ZeroAcquireTimeout,
262 #[error("datasource connect_timeout_ms must be greater than zero")]
264 ZeroConnectTimeout,
265}
266
267impl DataSourcePoolConfig {
268 pub fn validate_common(&self) -> Result<(), PoolConfigError> {
274 if self.url.trim().is_empty() {
275 return Err(PoolConfigError::EmptyUrl);
276 }
277 if self.max_connections == 0 {
278 return Err(PoolConfigError::ZeroMaxConnections);
279 }
280 if self.min_connections > self.max_connections {
281 return Err(PoolConfigError::InvalidConnectionRange);
282 }
283 if self.acquire_timeout_ms == 0 {
284 return Err(PoolConfigError::ZeroAcquireTimeout);
285 }
286 if self.connect_timeout_ms == 0 {
287 return Err(PoolConfigError::ZeroConnectTimeout);
288 }
289 Ok(())
290 }
291
292 pub fn endpoint(&self) -> String {
298 redacted_endpoint(&self.url)
299 }
300}
301
302pub fn redact_url(url: &str) -> String {
308 let Some((scheme, rest)) = url.split_once("://") else {
309 return "<redacted-url>".to_owned();
310 };
311 if scheme.is_empty()
312 || !scheme
313 .bytes()
314 .all(|value| value.is_ascii_alphanumeric() || matches!(value, b'+' | b'-' | b'.'))
315 {
316 return "<redacted-url>".to_owned();
317 }
318 let endpoint = redacted_endpoint(url);
319 if endpoint == "<unknown-endpoint>" {
320 return "<redacted-url>".to_owned();
321 }
322 let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
323 if authority.contains('@') {
324 format!("{scheme}://***@{endpoint}")
325 } else {
326 format!("{scheme}://{endpoint}")
327 }
328}
329
330pub fn redacted_endpoint(url: &str) -> String {
336 let Some((_, rest)) = url.split_once("://") else {
337 return "<unknown-endpoint>".to_owned();
338 };
339 let authority_and_path = rest.rsplit_once('@').map_or(rest, |(_, value)| value);
340 let endpoint = authority_and_path
341 .split(['?', '#'])
342 .next()
343 .unwrap_or(authority_and_path);
344 if endpoint.is_empty() {
345 "<unknown-endpoint>".to_owned()
346 } else {
347 endpoint.to_owned()
348 }
349}
350
351fn default_max_connections() -> u32 {
357 10
358}
359
360fn default_acquire_timeout_ms() -> u64 {
366 2_000
367}
368
369fn default_connect_timeout_ms() -> u64 {
375 5_000
376}
377
378fn default_probe_on_start() -> bool {
384 true
385}
386
387static NEXT_OWNER_ID: AtomicU64 = AtomicU64::new(1);
388
389#[derive(Debug)]
390struct OwnerIdentity {
392 id: u64,
393}
394
395#[derive(Clone)]
397pub struct ManagedRegistryOwner(Arc<OwnerIdentity>);
398
399impl ManagedRegistryOwner {
400 pub fn new() -> Self {
406 Self(Arc::new(OwnerIdentity {
407 id: NEXT_OWNER_ID.fetch_add(1, Ordering::Relaxed),
408 }))
409 }
410
411 pub fn ptr_eq(&self, other: &Self) -> bool {
417 Arc::ptr_eq(&self.0, &other.0)
418 }
419
420 pub fn diagnostic_id(&self) -> u64 {
426 self.0.id
427 }
428
429 fn downgrade(&self) -> Weak<OwnerIdentity> {
435 Arc::downgrade(&self.0)
436 }
437}
438
439impl Default for ManagedRegistryOwner {
440 fn default() -> Self {
446 Self::new()
447 }
448}
449
450impl std::fmt::Debug for ManagedRegistryOwner {
451 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
457 formatter
458 .debug_tuple("ManagedRegistryOwner")
459 .field(&self.diagnostic_id())
460 .finish()
461 }
462}
463
464#[derive(Debug, Clone, Copy, PartialEq, Eq)]
466pub enum BootstrapKind {
467 Configured,
469 DeferredDefault,
471}
472
473#[derive(Clone, Debug)]
475pub struct ManagedInstallationToken {
476 owner: ManagedRegistryOwner,
477 kind: BootstrapKind,
478}
479
480impl ManagedInstallationToken {
481 pub fn owner(&self) -> &ManagedRegistryOwner {
487 &self.owner
488 }
489
490 pub fn kind(&self) -> BootstrapKind {
496 self.kind
497 }
498}
499
500pub struct DataSourceCatalog {
502 owner: ManagedRegistryOwner,
503 entries: BTreeMap<DatasourceRef, DatabaseDriver>,
504 lifecycle: AtomicU8,
505}
506
507const CATALOG_STAGING: u8 = 0;
508const CATALOG_OPEN: u8 = 1;
509const CATALOG_STOPPED: u8 = 2;
510
511impl DataSourceCatalog {
512 pub fn try_new(
520 owner: ManagedRegistryOwner,
521 entries: impl IntoIterator<Item = (String, DatabaseDriver)>,
522 ) -> Result<Self, CatalogBuildError> {
523 let mut catalog = BTreeMap::new();
524 for (name, driver) in entries {
525 if catalog.len() >= MAX_MANAGED_DATASOURCES {
526 return Err(CatalogBuildError::TooManyDatasources);
527 }
528 let reference = DatasourceRef::try_new(name).map_err(CatalogBuildError::InvalidName)?;
529 if catalog.insert(reference.clone(), driver).is_some() {
530 return Err(CatalogBuildError::Duplicate(reference));
531 }
532 }
533 if catalog.is_empty() {
534 return Err(CatalogBuildError::Empty);
535 }
536 Ok(Self {
537 owner,
538 entries: catalog,
539 lifecycle: AtomicU8::new(CATALOG_STAGING),
540 })
541 }
542
543 pub fn owner(&self) -> &ManagedRegistryOwner {
549 &self.owner
550 }
551
552 pub fn len(&self) -> usize {
558 self.entries.len()
559 }
560
561 pub fn is_empty(&self) -> bool {
567 self.entries.is_empty()
568 }
569
570 pub fn entries(&self) -> Vec<(DatasourceRef, DatabaseDriver)> {
576 self.entries
577 .iter()
578 .map(|(reference, driver)| (reference.clone(), *driver))
579 .collect()
580 }
581
582 pub fn resolve(
590 &self,
591 datasource: &str,
592 expected: DatabaseDriver,
593 ) -> Result<DatasourceRef, DataSourceLookupError> {
594 match self.lifecycle.load(Ordering::Acquire) {
595 CATALOG_OPEN => {}
596 CATALOG_STOPPED => return Err(DataSourceLookupError::Stopped),
597 _ => return Err(DataSourceLookupError::RegistryUnavailable),
598 }
599 let reference =
600 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)?;
601 let actual = self.entries.get(&reference).copied().ok_or_else(|| {
602 DataSourceLookupError::NotFound {
603 datasource: reference.clone(),
604 }
605 })?;
606 if actual != expected {
607 return Err(DataSourceLookupError::DriverMismatch {
608 datasource: reference,
609 expected,
610 actual,
611 });
612 }
613 Ok(reference)
614 }
615
616 fn open(&self) {
622 self.lifecycle.store(CATALOG_OPEN, Ordering::Release);
623 }
624
625 fn stop_accepting(&self) {
631 self.lifecycle.store(CATALOG_STOPPED, Ordering::Release);
632 }
633}
634
635#[derive(Debug, thiserror::Error)]
637pub enum CatalogBuildError {
638 #[error("managed datasource catalog cannot be empty")]
640 Empty,
641 #[error("managed datasource count exceeds the supported limit")]
643 TooManyDatasources,
644 #[error("invalid datasource name")]
646 InvalidName(#[source] DatasourceNameError),
647 #[error("datasource `{0}` is configured more than once")]
649 Duplicate(DatasourceRef),
650}
651
652#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
654pub enum DataSourceLookupError {
655 #[error("invalid datasource name")]
657 InvalidName,
658 #[error("datasource registry is unavailable")]
660 RegistryUnavailable,
661 #[error("datasource registry is stopped")]
663 Stopped,
664 #[error("datasource `{datasource}` is not managed by this application")]
666 NotFound {
667 datasource: DatasourceRef,
669 },
670 #[error("datasource `{datasource}` uses {actual}, not {expected}")]
672 DriverMismatch {
673 datasource: DatasourceRef,
675 expected: DatabaseDriver,
677 actual: DatabaseDriver,
679 },
680}
681
682#[must_use]
684#[derive(Debug, Clone, PartialEq, Eq)]
685pub enum TxDecision<T, E> {
686 Commit(T),
688 Rollback(E),
690}
691
692#[derive(Debug)]
694pub enum TxRollbackCause<E> {
695 Decision(E),
697 RollbackOnly {
699 reason: String,
701 },
702}
703
704#[derive(Debug)]
706pub enum TxRunError<E> {
707 Rollback(E),
709 RollbackOnly {
711 reason: String,
713 },
714 CommitRejected {
716 reason: String,
718 },
719 CommitUncertain {
721 reason: String,
723 },
724 RollbackFailed {
726 cause: TxRollbackCause<E>,
728 reason: String,
730 },
731 Infrastructure {
733 reason: String,
735 },
736}
737
738#[derive(Debug)]
740pub enum TxEntryError<E> {
741 Lookup(DataSourceLookupError),
743 Run(TxRunError<E>),
745}
746
747impl<E: std::fmt::Display> std::fmt::Display for TxEntryError<E> {
748 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
754 match self {
755 Self::Lookup(error) => {
756 write!(formatter, "transaction datasource lookup failed: {error}")
757 }
758 Self::Run(error) => std::fmt::Display::fmt(error, formatter),
759 }
760 }
761}
762
763impl<E> std::error::Error for TxEntryError<E> where E: std::fmt::Debug + std::fmt::Display {}
764
765#[derive(Debug, Clone, Copy, PartialEq, Eq)]
767pub enum CommitOutcome {
768 Committed,
770 Rejected,
772 OutcomeUnknown,
774}
775
776impl<E: std::fmt::Display> std::fmt::Display for TxRunError<E> {
777 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
783 match self {
784 Self::Rollback(error) => write!(formatter, "transaction rolled back: {error}"),
785 Self::RollbackOnly { reason } => {
786 write!(formatter, "transaction was rollback-only: {reason}")
787 }
788 Self::CommitRejected { reason } => write!(formatter, "commit rejected: {reason}"),
789 Self::CommitUncertain { reason } => write!(formatter, "commit uncertain: {reason}"),
790 Self::RollbackFailed { reason, .. } => write!(formatter, "rollback failed: {reason}"),
791 Self::Infrastructure { reason } => {
792 write!(formatter, "transaction infrastructure failed: {reason}")
793 }
794 }
795 }
796}
797
798impl<E> std::error::Error for TxRunError<E> where E: std::fmt::Debug + std::fmt::Display {}
799
800#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
802#[error("ambient transaction driver mismatch: active={active}, requested={requested}")]
803pub struct DriverScopeError {
804 pub active: DatabaseDriver,
806 pub requested: DatabaseDriver,
808}
809
810tokio::task_local! {
811 static ACTIVE_DRIVER: DatabaseDriver;
812}
813
814pub fn current_driver() -> Option<DatabaseDriver> {
820 ACTIVE_DRIVER.try_with(|driver| *driver).ok()
821}
822
823pub fn ensure_driver(requested: DatabaseDriver) -> Result<(), DriverScopeError> {
829 match current_driver() {
830 Some(active) if active != requested => Err(DriverScopeError { active, requested }),
831 _ => Ok(()),
832 }
833}
834
835pub async fn scope_driver<F>(
843 driver: DatabaseDriver,
844 future: F,
845) -> Result<F::Output, DriverScopeError>
846where
847 F: Future,
848{
849 match current_driver() {
850 Some(active) if active != driver => Err(DriverScopeError {
851 active,
852 requested: driver,
853 }),
854 Some(_) => Ok(future.await),
855 None => Ok(ACTIVE_DRIVER.scope(driver, future).await),
856 }
857}
858
859#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
861pub enum RegistryCoordinationError {
862 #[error("standalone datasource driver conflicts with the active driver")]
864 StandaloneDriverConflict,
865 #[error("standalone datasource registration conflicts with managed application state")]
867 ManagedModeConflict,
868 #[error("deferred datasource bootstrap accepts only the default datasource")]
870 DeferredOnlyDefault,
871 #[error("deferred datasource bootstrap already uses another driver")]
873 DeferredDriverConflict,
874 #[error("datasource registry is already occupied")]
876 RegistryOccupied,
877 #[error("managed registry owner token does not match the active bootstrap")]
879 OwnerMismatch,
880 #[error("datasource catalog owner does not match the installation token")]
882 CatalogOwnerMismatch,
883 #[error("datasource catalog identity does not match the active managed catalog")]
885 CatalogIdentityMismatch,
886 #[error("deferred datasource bootstrap did not install the expected default driver")]
888 DeferredDriverMissing,
889 #[error("deferred datasource resources must be removed before bootstrap cleanup")]
891 DeferredResourcesRemain,
892 #[error("public standalone adopt is unavailable during application bootstrap")]
894 PublicAdoptDuringBootstrap,
895 #[error("orphaned deferred datasource resources require explicit cleanup")]
897 OrphanedDeferredResources,
898 #[error("deferred datasource catalog must contain only the installed default driver")]
900 InvalidDeferredCatalog,
901}
902
903#[derive(Debug)]
904enum RegistryMode {
905 Empty,
906 Standalone(DatabaseDriver),
907 Bootstrapping {
908 owner: Weak<OwnerIdentity>,
909 kind: BootstrapKind,
910 deferred_driver: Option<DatabaseDriver>,
911 },
912 Managed {
913 owner: Weak<OwnerIdentity>,
914 catalog: Weak<DataSourceCatalog>,
915 },
916}
917
918#[derive(Debug)]
919struct RegistryState {
921 mode: RegistryMode,
922}
923
924impl Default for RegistryState {
925 fn default() -> Self {
931 Self {
932 mode: RegistryMode::Empty,
933 }
934 }
935}
936
937static REGISTRY_STATE: OnceLock<Mutex<RegistryState>> = OnceLock::new();
938
939#[derive(Debug, Clone, Copy, PartialEq, Eq)]
941pub enum RegistryModeSnapshot {
942 Empty,
944 Standalone(DatabaseDriver),
946 Bootstrapping {
948 kind: BootstrapKind,
950 deferred_driver: Option<DatabaseDriver>,
952 },
953 Managed,
955 OrphanedDeferred(DatabaseDriver),
957}
958
959#[doc(hidden)]
965pub fn coordinate<T>(operation: impl FnOnce(&mut RegistryCoordinator<'_>) -> T) -> T {
966 let mut state = REGISTRY_STATE
967 .get_or_init(|| Mutex::new(RegistryState::default()))
968 .lock()
969 .unwrap_or_else(|error| error.into_inner());
970 let mut coordinator = RegistryCoordinator { state: &mut state };
971 coordinator.sanitize();
972 operation(&mut coordinator)
973}
974
975#[doc(hidden)]
977pub struct RegistryCoordinator<'a> {
978 state: &'a mut RegistryState,
979}
980
981impl RegistryCoordinator<'_> {
982 fn sanitize(&mut self) {
988 match &self.state.mode {
989 RegistryMode::Bootstrapping {
990 owner,
991 deferred_driver: None,
992 ..
993 } if owner.upgrade().is_none() => {
994 self.state.mode = RegistryMode::Empty;
995 }
996 RegistryMode::Managed { owner, catalog }
997 if owner.upgrade().is_none() && catalog.upgrade().is_none() =>
998 {
999 self.state.mode = RegistryMode::Empty;
1000 }
1001 _ => {}
1002 }
1003 }
1004
1005 pub fn register_standalone(
1013 &mut self,
1014 driver: DatabaseDriver,
1015 datasource: &str,
1016 ) -> Result<(), RegistryCoordinationError> {
1017 match &mut self.state.mode {
1018 RegistryMode::Empty => {
1019 self.state.mode = RegistryMode::Standalone(driver);
1020 Ok(())
1021 }
1022 RegistryMode::Standalone(active) if *active == driver => Ok(()),
1023 RegistryMode::Standalone(_) => Err(RegistryCoordinationError::StandaloneDriverConflict),
1024 RegistryMode::Bootstrapping {
1025 owner,
1026 kind: BootstrapKind::DeferredDefault,
1027 deferred_driver,
1028 } => {
1029 if owner.upgrade().is_none() && deferred_driver.is_some() {
1030 return Err(RegistryCoordinationError::OrphanedDeferredResources);
1031 }
1032 if datasource != DEFAULT_DATASOURCE {
1033 return Err(RegistryCoordinationError::DeferredOnlyDefault);
1034 }
1035 match deferred_driver {
1036 Some(active) if *active != driver => {
1037 Err(RegistryCoordinationError::DeferredDriverConflict)
1038 }
1039 Some(_) => Ok(()),
1040 slot @ None => {
1041 *slot = Some(driver);
1042 Ok(())
1043 }
1044 }
1045 }
1046 RegistryMode::Bootstrapping { .. } | RegistryMode::Managed { .. } => {
1047 Err(RegistryCoordinationError::ManagedModeConflict)
1048 }
1049 }
1050 }
1051
1052 pub fn release_standalone(
1058 &mut self,
1059 driver: DatabaseDriver,
1060 ) -> Result<(), RegistryCoordinationError> {
1061 match &mut self.state.mode {
1062 RegistryMode::Standalone(active) if *active == driver => {
1063 self.state.mode = RegistryMode::Empty;
1064 Ok(())
1065 }
1066 RegistryMode::Bootstrapping {
1067 owner,
1068 kind: BootstrapKind::DeferredDefault,
1069 deferred_driver: Some(active),
1070 } if *active == driver => {
1071 let owner_alive = owner.upgrade().is_some();
1072 if owner_alive {
1073 if let RegistryMode::Bootstrapping {
1074 deferred_driver, ..
1075 } = &mut self.state.mode
1076 {
1077 *deferred_driver = None;
1078 }
1079 } else {
1080 self.state.mode = RegistryMode::Empty;
1081 }
1082 Ok(())
1083 }
1084 RegistryMode::Empty => Ok(()),
1085 _ => Err(RegistryCoordinationError::ManagedModeConflict),
1086 }
1087 }
1088
1089 pub fn begin_bootstrap(
1097 &mut self,
1098 owner: &ManagedRegistryOwner,
1099 kind: BootstrapKind,
1100 ) -> Result<ManagedInstallationToken, RegistryCoordinationError> {
1101 if !matches!(self.state.mode, RegistryMode::Empty) {
1102 return Err(RegistryCoordinationError::RegistryOccupied);
1103 }
1104 self.state.mode = RegistryMode::Bootstrapping {
1105 owner: owner.downgrade(),
1106 kind,
1107 deferred_driver: None,
1108 };
1109 Ok(ManagedInstallationToken {
1110 owner: owner.clone(),
1111 kind,
1112 })
1113 }
1114
1115 pub fn begin_legacy_adopt(
1123 &mut self,
1124 owner: &ManagedRegistryOwner,
1125 driver: DatabaseDriver,
1126 ) -> Result<ManagedInstallationToken, RegistryCoordinationError> {
1127 match &self.state.mode {
1128 RegistryMode::Standalone(active) if *active == driver => {
1129 self.state.mode = RegistryMode::Bootstrapping {
1130 owner: owner.downgrade(),
1131 kind: BootstrapKind::Configured,
1132 deferred_driver: None,
1133 };
1134 Ok(ManagedInstallationToken {
1135 owner: owner.clone(),
1136 kind: BootstrapKind::Configured,
1137 })
1138 }
1139 RegistryMode::Bootstrapping { .. } => {
1140 Err(RegistryCoordinationError::PublicAdoptDuringBootstrap)
1141 }
1142 _ => Err(RegistryCoordinationError::RegistryOccupied),
1143 }
1144 }
1145
1146 pub fn restore_legacy_standalone(
1154 &mut self,
1155 token: &ManagedInstallationToken,
1156 driver: DatabaseDriver,
1157 ) -> Result<(), RegistryCoordinationError> {
1158 self.verify_bootstrap_owner(token)?;
1159 self.state.mode = RegistryMode::Standalone(driver);
1160 Ok(())
1161 }
1162
1163 pub fn verify_deferred_adopt(
1171 &mut self,
1172 token: &ManagedInstallationToken,
1173 driver: DatabaseDriver,
1174 ) -> Result<(), RegistryCoordinationError> {
1175 self.verify_bootstrap_owner(token)?;
1176 match &self.state.mode {
1177 RegistryMode::Bootstrapping {
1178 kind: BootstrapKind::DeferredDefault,
1179 deferred_driver: Some(active),
1180 ..
1181 } if *active == driver => Ok(()),
1182 _ => Err(RegistryCoordinationError::DeferredDriverMissing),
1183 }
1184 }
1185
1186 pub fn verify_installation_token(
1192 &self,
1193 token: &ManagedInstallationToken,
1194 ) -> Result<(), RegistryCoordinationError> {
1195 self.verify_bootstrap_owner(token)
1196 }
1197
1198 pub fn install_catalog(
1206 &mut self,
1207 token: &ManagedInstallationToken,
1208 catalog: &Arc<DataSourceCatalog>,
1209 ) -> Result<(), RegistryCoordinationError> {
1210 self.verify_bootstrap_owner(token)?;
1211 if !catalog.owner().ptr_eq(token.owner()) {
1212 return Err(RegistryCoordinationError::CatalogOwnerMismatch);
1213 }
1214 if let RegistryMode::Bootstrapping {
1215 kind: BootstrapKind::DeferredDefault,
1216 deferred_driver: Some(driver),
1217 ..
1218 } = &self.state.mode
1219 {
1220 let entries = catalog.entries();
1221 if entries.len() != 1
1222 || entries[0].0.as_str() != DEFAULT_DATASOURCE
1223 || entries[0].1 != *driver
1224 {
1225 return Err(RegistryCoordinationError::InvalidDeferredCatalog);
1226 }
1227 }
1228 self.state.mode = RegistryMode::Managed {
1229 owner: token.owner.downgrade(),
1230 catalog: Arc::downgrade(catalog),
1231 };
1232 Ok(())
1233 }
1234
1235 pub fn open_catalog(
1243 &mut self,
1244 token: &ManagedInstallationToken,
1245 catalog: &Arc<DataSourceCatalog>,
1246 ) -> Result<(), RegistryCoordinationError> {
1247 match &self.state.mode {
1248 RegistryMode::Managed {
1249 owner,
1250 catalog: active,
1251 } => {
1252 let Some(active_owner) = owner.upgrade() else {
1253 return Err(RegistryCoordinationError::OwnerMismatch);
1254 };
1255 if !Arc::ptr_eq(&active_owner, &token.owner.0) {
1256 return Err(RegistryCoordinationError::OwnerMismatch);
1257 }
1258 let Some(active_catalog) = active.upgrade() else {
1259 return Err(RegistryCoordinationError::CatalogIdentityMismatch);
1260 };
1261 if !Arc::ptr_eq(&active_catalog, catalog) {
1262 return Err(RegistryCoordinationError::CatalogIdentityMismatch);
1263 }
1264 catalog.open();
1265 Ok(())
1266 }
1267 _ => Err(RegistryCoordinationError::OwnerMismatch),
1268 }
1269 }
1270
1271 pub fn clear_managed(
1277 &mut self,
1278 owner: &ManagedRegistryOwner,
1279 ) -> Result<(), RegistryCoordinationError> {
1280 match &self.state.mode {
1281 RegistryMode::Managed {
1282 owner: active,
1283 catalog,
1284 } => {
1285 let Some(active_owner) = active.upgrade() else {
1286 return Err(RegistryCoordinationError::OwnerMismatch);
1287 };
1288 if !Arc::ptr_eq(&active_owner, &owner.0) {
1289 return Err(RegistryCoordinationError::OwnerMismatch);
1290 }
1291 if let Some(catalog) = catalog.upgrade() {
1292 catalog.stop_accepting();
1293 }
1294 self.state.mode = RegistryMode::Empty;
1295 Ok(())
1296 }
1297 _ => Err(RegistryCoordinationError::OwnerMismatch),
1298 }
1299 }
1300
1301 pub fn clear_deferred_driver(
1309 &mut self,
1310 token: &ManagedInstallationToken,
1311 driver: DatabaseDriver,
1312 ) -> Result<(), RegistryCoordinationError> {
1313 self.verify_bootstrap_owner(token)?;
1314 match &mut self.state.mode {
1315 RegistryMode::Bootstrapping {
1316 kind: BootstrapKind::DeferredDefault,
1317 deferred_driver: Some(active),
1318 ..
1319 } if *active == driver => {
1320 if let RegistryMode::Bootstrapping {
1321 deferred_driver, ..
1322 } = &mut self.state.mode
1323 {
1324 *deferred_driver = None;
1325 }
1326 Ok(())
1327 }
1328 _ => Err(RegistryCoordinationError::DeferredDriverMissing),
1329 }
1330 }
1331
1332 pub fn abort_bootstrap(
1338 &mut self,
1339 token: &ManagedInstallationToken,
1340 ) -> Result<(), RegistryCoordinationError> {
1341 self.verify_bootstrap_owner(token)?;
1342 match &self.state.mode {
1343 RegistryMode::Bootstrapping {
1344 deferred_driver: Some(_),
1345 ..
1346 } => Err(RegistryCoordinationError::DeferredResourcesRemain),
1347 RegistryMode::Bootstrapping { .. } => {
1348 self.state.mode = RegistryMode::Empty;
1349 Ok(())
1350 }
1351 _ => Err(RegistryCoordinationError::OwnerMismatch),
1352 }
1353 }
1354
1355 pub fn snapshot(&self) -> RegistryModeSnapshot {
1361 match &self.state.mode {
1362 RegistryMode::Empty => RegistryModeSnapshot::Empty,
1363 RegistryMode::Standalone(driver) => RegistryModeSnapshot::Standalone(*driver),
1364 RegistryMode::Bootstrapping {
1365 owner,
1366 kind,
1367 deferred_driver,
1368 } => match (owner.upgrade(), deferred_driver) {
1369 (None, Some(driver)) => RegistryModeSnapshot::OrphanedDeferred(*driver),
1370 _ => RegistryModeSnapshot::Bootstrapping {
1371 kind: *kind,
1372 deferred_driver: *deferred_driver,
1373 },
1374 },
1375 RegistryMode::Managed { .. } => RegistryModeSnapshot::Managed,
1376 }
1377 }
1378
1379 fn verify_bootstrap_owner(
1385 &self,
1386 token: &ManagedInstallationToken,
1387 ) -> Result<(), RegistryCoordinationError> {
1388 match &self.state.mode {
1389 RegistryMode::Bootstrapping { owner, kind, .. } => {
1390 let Some(active) = owner.upgrade() else {
1391 return Err(RegistryCoordinationError::OwnerMismatch);
1392 };
1393 if Arc::ptr_eq(&active, &token.owner.0) && *kind == token.kind {
1394 Ok(())
1395 } else {
1396 Err(RegistryCoordinationError::OwnerMismatch)
1397 }
1398 }
1399 _ => Err(RegistryCoordinationError::OwnerMismatch),
1400 }
1401 }
1402}
1403
1404pub fn registry_mode() -> RegistryModeSnapshot {
1410 coordinate(|coordinator| coordinator.snapshot())
1411}
1412
1413pub fn begin_managed_bootstrap(
1419 kind: BootstrapKind,
1420) -> Result<(ManagedRegistryOwner, ManagedInstallationToken), RegistryCoordinationError> {
1421 let owner = ManagedRegistryOwner::new();
1422 let token = coordinate(|coordinator| coordinator.begin_bootstrap(&owner, kind))?;
1423 Ok((owner, token))
1424}
1425
1426pub fn install_managed_catalog(
1434 token: &ManagedInstallationToken,
1435 catalog: &Arc<DataSourceCatalog>,
1436) -> Result<(), RegistryCoordinationError> {
1437 coordinate(|coordinator| coordinator.install_catalog(token, catalog))
1438}
1439
1440pub fn open_managed_catalog(
1448 token: &ManagedInstallationToken,
1449 catalog: &Arc<DataSourceCatalog>,
1450) -> Result<(), RegistryCoordinationError> {
1451 coordinate(|coordinator| coordinator.open_catalog(token, catalog))
1452}
1453
1454pub fn clear_managed_catalog(
1460 owner: &ManagedRegistryOwner,
1461) -> Result<(), RegistryCoordinationError> {
1462 coordinate(|coordinator| coordinator.clear_managed(owner))
1463}
1464
1465pub fn abort_managed_bootstrap(
1471 token: &ManagedInstallationToken,
1472) -> Result<(), RegistryCoordinationError> {
1473 coordinate(|coordinator| coordinator.abort_bootstrap(token))
1474}
1475
1476pub fn resolve_datasource(
1485 datasource: &str,
1486 expected: DatabaseDriver,
1487) -> Result<DatasourceRef, DataSourceLookupError> {
1488 coordinate(|coordinator| match &coordinator.state.mode {
1489 RegistryMode::Empty => {
1490 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
1493 }
1494 RegistryMode::Bootstrapping {
1495 kind: BootstrapKind::DeferredDefault,
1496 deferred_driver: Some(actual),
1497 ..
1498 } if *actual == expected && datasource == DEFAULT_DATASOURCE => {
1499 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
1502 }
1503 RegistryMode::Bootstrapping { .. } => Err(DataSourceLookupError::RegistryUnavailable),
1504 RegistryMode::Standalone(actual) if *actual == expected => {
1505 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)
1506 }
1507 RegistryMode::Standalone(actual) => {
1508 let reference =
1509 DatasourceRef::new(datasource).map_err(|_| DataSourceLookupError::InvalidName)?;
1510 Err(DataSourceLookupError::DriverMismatch {
1511 datasource: reference,
1512 expected,
1513 actual: *actual,
1514 })
1515 }
1516 RegistryMode::Managed { catalog, .. } => catalog
1517 .upgrade()
1518 .ok_or(DataSourceLookupError::RegistryUnavailable)?
1519 .resolve(datasource, expected),
1520 })
1521}
1522
1523pub fn managed_catalog() -> Result<Arc<DataSourceCatalog>, DataSourceLookupError> {
1529 coordinate(|coordinator| match &coordinator.state.mode {
1530 RegistryMode::Managed { catalog, .. } => catalog
1531 .upgrade()
1532 .ok_or(DataSourceLookupError::RegistryUnavailable)
1533 .and_then(|catalog| match catalog.lifecycle.load(Ordering::Acquire) {
1534 CATALOG_OPEN => Ok(catalog),
1535 CATALOG_STOPPED => Err(DataSourceLookupError::Stopped),
1536 _ => Err(DataSourceLookupError::RegistryUnavailable),
1537 }),
1538 _ => Err(DataSourceLookupError::RegistryUnavailable),
1539 })
1540}