state_repository.rs (18886B)
1 //! Sealed typed access to Myc-owned SQLite state. 2 3 use core::fmt; 4 use std::error::Error; 5 6 use radroots_service_sqlite::{ 7 ServiceSqliteHost, ServiceSqliteTransaction, ServiceSqliteTransactionError, 8 ServiceSqliteTransactionErrorKind, 9 }; 10 use sqlx::Row; 11 12 use crate::{ 13 MYC_STATE_SCHEMA_VERSION, MycConfigDocumentV1, MycStateMetadata, 14 state_metadata::{expected_identities, normalized_config_digest}, 15 }; 16 17 const READ_METADATA_SQL: &str = r#"SELECT 18 singleton, 19 CASE 20 WHEN typeof(normalized_config_sha256) = 'blob' 21 AND length(normalized_config_sha256) = 32 22 THEN normalized_config_sha256 23 ELSE NULL 24 END AS normalized_config_sha256, 25 CASE 26 WHEN typeof(transport_public_key) = 'text' 27 AND length(CAST(transport_public_key AS BLOB)) = 64 28 THEN transport_public_key 29 ELSE NULL 30 END AS transport_public_key, 31 CASE 32 WHEN typeof(user_public_key) = 'text' 33 AND length(CAST(user_public_key AS BLOB)) = 64 34 THEN user_public_key 35 ELSE NULL 36 END AS user_public_key, 37 typeof(discovery_public_key) AS discovery_public_key_type, 38 CASE 39 WHEN typeof(discovery_public_key) = 'text' 40 AND length(CAST(discovery_public_key AS BLOB)) = 64 41 THEN discovery_public_key 42 ELSE NULL 43 END AS discovery_public_key, 44 config_contract_version, 45 state_contract_version, 46 operator_contract_version, 47 status_contract_version 48 FROM myc_state_metadata 49 LIMIT 2"#; 50 51 const INSERT_METADATA_SQL: &str = r#"INSERT INTO myc_state_metadata ( 52 singleton, 53 normalized_config_sha256, 54 transport_public_key, 55 user_public_key, 56 discovery_public_key, 57 config_contract_version, 58 state_contract_version, 59 operator_contract_version, 60 status_contract_version 61 ) VALUES (1, ?, ?, ?, ?, ?, ?, ?, ?)"#; 62 63 const READ_LATEST_CONFIG_BINDING_SQL: &str = r#"SELECT 64 CASE 65 WHEN typeof(normalized_config_sha256) = 'blob' 66 AND length(normalized_config_sha256) = 32 67 THEN normalized_config_sha256 68 ELSE NULL 69 END AS normalized_config_sha256, 70 CASE 71 WHEN typeof(transport_public_key) = 'text' 72 AND length(CAST(transport_public_key AS BLOB)) = 64 73 THEN transport_public_key 74 ELSE NULL 75 END AS transport_public_key, 76 CASE 77 WHEN typeof(user_public_key) = 'text' 78 AND length(CAST(user_public_key AS BLOB)) = 64 79 THEN user_public_key 80 ELSE NULL 81 END AS user_public_key, 82 typeof(discovery_public_key) AS discovery_public_key_type, 83 CASE 84 WHEN typeof(discovery_public_key) = 'text' 85 AND length(CAST(discovery_public_key AS BLOB)) = 64 86 THEN discovery_public_key 87 ELSE NULL 88 END AS discovery_public_key, 89 config_contract_version, 90 state_contract_version, 91 operator_contract_version, 92 status_contract_version 93 FROM myc_config_bindings 94 ORDER BY generation DESC 95 LIMIT 1"#; 96 97 const INSERT_INITIAL_CONFIG_BINDING_SQL: &str = r#"INSERT INTO myc_config_bindings ( 98 generation, normalized_config_sha256, transport_public_key, user_public_key, 99 discovery_public_key, config_contract_version, state_contract_version, 100 operator_contract_version, status_contract_version, applied_at_unix_s, 101 service_version, service_commit, lib_revision, rust_version, target, 102 feature_profile, provider_contract_version 103 ) 104 SELECT 1, metadata.normalized_config_sha256, metadata.transport_public_key, 105 metadata.user_public_key, metadata.discovery_public_key, 106 metadata.config_contract_version, ?, 107 metadata.operator_contract_version, metadata.status_contract_version, 108 migration.applied_at_unix_s, migration.service_version, 109 migration.service_commit, migration.lib_revision, migration.rust_version, 110 migration.target, migration.feature_profile, migration.provider_contract_version 111 FROM myc_state_metadata AS metadata 112 JOIN schema_migrations AS migration ON migration.version = ? 113 WHERE metadata.singleton = 1"#; 114 115 /// Stable failure classes for typed Myc state-repository operations. 116 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 117 pub enum MycStateRepositoryErrorKind { 118 Binding, 119 Transaction, 120 CommitOutcomeUnknown, 121 } 122 123 impl MycStateRepositoryErrorKind { 124 /// Returns the stable machine-readable failure code. 125 #[must_use] 126 pub const fn code(self) -> &'static str { 127 match self { 128 Self::Binding => "state_repository_binding_invalid", 129 Self::Transaction => "state_repository_transaction_failed", 130 Self::CommitOutcomeUnknown => "state_repository_commit_outcome_unknown", 131 } 132 } 133 } 134 135 /// Source-free typed repository failure. 136 #[derive(Clone, Copy, PartialEq, Eq)] 137 pub struct MycStateRepositoryError { 138 kind: MycStateRepositoryErrorKind, 139 } 140 141 impl MycStateRepositoryError { 142 pub(crate) const fn new(kind: MycStateRepositoryErrorKind) -> Self { 143 Self { kind } 144 } 145 146 /// Returns the stable failure class. 147 #[must_use] 148 pub const fn kind(self) -> MycStateRepositoryErrorKind { 149 self.kind 150 } 151 152 /// Returns the stable machine-readable failure code. 153 #[must_use] 154 pub const fn code(self) -> &'static str { 155 self.kind.code() 156 } 157 } 158 159 impl fmt::Display for MycStateRepositoryError { 160 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 161 formatter.write_str(match self.kind { 162 MycStateRepositoryErrorKind::Binding => "Myc state repository binding is invalid", 163 MycStateRepositoryErrorKind::Transaction => "Myc state repository transaction failed", 164 MycStateRepositoryErrorKind::CommitOutcomeUnknown => { 165 "Myc state repository commit outcome is unknown" 166 } 167 }) 168 } 169 } 170 171 impl fmt::Debug for MycStateRepositoryError { 172 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 173 formatter 174 .debug_struct("MycStateRepositoryError") 175 .field("kind", &self.kind) 176 .finish() 177 } 178 } 179 180 impl Error for MycStateRepositoryError {} 181 182 /// Borrowed typed access to one already-opened Myc state host. 183 /// 184 /// Construction is sealed to [`crate::MycStateHost::repository`]: 185 /// 186 /// ```compile_fail 187 /// use myc::MycStateRepository; 188 /// 189 /// let _ = MycStateRepository { host: todo!(), expected: todo!() }; 190 /// ``` 191 pub struct MycStateRepository<'host> { 192 host: &'host ServiceSqliteHost, 193 expected: &'host MycStateMetadata, 194 writable: bool, 195 } 196 197 impl<'host> MycStateRepository<'host> { 198 pub(crate) const fn new( 199 host: &'host ServiceSqliteHost, 200 expected: &'host MycStateMetadata, 201 writable: bool, 202 ) -> Self { 203 Self { 204 host, 205 expected, 206 writable, 207 } 208 } 209 210 pub(crate) const fn host(&self) -> &'host ServiceSqliteHost { 211 self.host 212 } 213 214 pub(crate) const fn expected(&self) -> &'host MycStateMetadata { 215 self.expected 216 } 217 218 pub(crate) const fn is_writable(&self) -> bool { 219 self.writable 220 } 221 222 /// Re-verifies the immutable Myc binding through the sealed transaction executor. 223 pub async fn verify_binding(&self) -> Result<(), MycStateRepositoryError> { 224 self.transact(false).await 225 } 226 227 pub(crate) async fn bind_or_verify(&self) -> Result<(), MycStateRepositoryError> { 228 self.transact(true).await 229 } 230 231 async fn transact(&self, initialize_missing: bool) -> Result<(), MycStateRepositoryError> { 232 let expected = PersistedMetadata::from(self.expected); 233 self.host 234 .transaction(move |transaction| { 235 Box::pin(async move { 236 let birth = read_metadata(transaction).await?; 237 match birth { 238 Some(actual) if actual.same_contracts(&expected) => {} 239 Some(_) => return Err(RepositoryOperationError::Binding), 240 None if initialize_missing => { 241 insert_metadata(transaction, &expected).await?; 242 match read_metadata(transaction).await? { 243 Some(actual) if actual == expected => {} 244 Some(_) | None => return Err(RepositoryOperationError::Binding), 245 } 246 } 247 None => return Err(RepositoryOperationError::Binding), 248 } 249 let latest = read_latest_config_binding(transaction).await?; 250 if latest.is_none() && initialize_missing { 251 insert_initial_config_binding(transaction).await?; 252 } 253 match read_latest_config_binding(transaction).await? { 254 Some(actual) if actual.matches_current_configuration(&expected) => Ok(()), 255 Some(_) | None => Err(RepositoryOperationError::Binding), 256 } 257 }) 258 }) 259 .await 260 .map_err(map_transaction_error) 261 } 262 } 263 264 impl fmt::Debug for MycStateRepository<'_> { 265 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 266 formatter 267 .debug_struct("MycStateRepository") 268 .field("state", &"[sealed]") 269 .finish() 270 } 271 } 272 273 #[derive(Clone, PartialEq, Eq)] 274 pub(crate) struct PersistedMetadata { 275 pub(crate) normalized_config_sha256: [u8; 32], 276 pub(crate) transport_public_key: Box<str>, 277 pub(crate) user_public_key: Box<str>, 278 pub(crate) discovery_public_key: Option<Box<str>>, 279 pub(crate) config_contract_version: u32, 280 pub(crate) state_contract_version: u32, 281 pub(crate) operator_contract_version: u32, 282 pub(crate) status_contract_version: u32, 283 } 284 285 impl PersistedMetadata { 286 pub(crate) fn from_configuration( 287 configuration: &MycConfigDocumentV1, 288 ) -> Result<Self, RepositoryOperationError> { 289 let identities = expected_identities(configuration.normalized()) 290 .map_err(|_| RepositoryOperationError::Binding)?; 291 let digest = normalized_config_digest(configuration.profile(), configuration.normalized()) 292 .map_err(|_| RepositoryOperationError::Binding)?; 293 Ok(Self { 294 normalized_config_sha256: *digest.as_bytes(), 295 transport_public_key: identities.transport().as_hex().into(), 296 user_public_key: identities.user().as_hex().into(), 297 discovery_public_key: identities.discovery().map(|value| value.as_hex().into()), 298 config_contract_version: configuration.schema_version(), 299 state_contract_version: MYC_STATE_SCHEMA_VERSION, 300 operator_contract_version: crate::MYC_OPERATOR_CONTRACT_VERSION, 301 status_contract_version: crate::MYC_SIGNER_STATUS_CONTRACT_VERSION, 302 }) 303 } 304 305 fn same_contracts(&self, other: &Self) -> bool { 306 self.config_contract_version == other.config_contract_version 307 && (9..=MYC_STATE_SCHEMA_VERSION).contains(&self.state_contract_version) 308 && other.state_contract_version == MYC_STATE_SCHEMA_VERSION 309 && self.operator_contract_version == other.operator_contract_version 310 && self.status_contract_version == other.status_contract_version 311 } 312 313 pub(crate) fn matches_current_configuration(&self, other: &Self) -> bool { 314 self.normalized_config_sha256 == other.normalized_config_sha256 315 && self.transport_public_key == other.transport_public_key 316 && self.user_public_key == other.user_public_key 317 && self.discovery_public_key == other.discovery_public_key 318 && self.config_contract_version == other.config_contract_version 319 && (10..=MYC_STATE_SCHEMA_VERSION).contains(&self.state_contract_version) 320 && other.state_contract_version == MYC_STATE_SCHEMA_VERSION 321 && self.operator_contract_version == other.operator_contract_version 322 && self.status_contract_version == other.status_contract_version 323 } 324 } 325 326 impl From<&MycStateMetadata> for PersistedMetadata { 327 fn from(metadata: &MycStateMetadata) -> Self { 328 let identities = metadata.expected_identities(); 329 let versions = metadata.policy_versions(); 330 Self { 331 normalized_config_sha256: *metadata.configuration_digest().as_bytes(), 332 transport_public_key: identities.transport().as_hex().into(), 333 user_public_key: identities.user().as_hex().into(), 334 discovery_public_key: identities.discovery().map(|value| value.as_hex().into()), 335 config_contract_version: versions.configuration(), 336 state_contract_version: versions.state(), 337 operator_contract_version: versions.operator(), 338 status_contract_version: versions.status(), 339 } 340 } 341 } 342 343 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 344 pub(crate) enum RepositoryOperationError { 345 Binding, 346 Storage, 347 } 348 349 pub(crate) async fn require_expected_metadata( 350 transaction: &mut ServiceSqliteTransaction<'_>, 351 expected: &PersistedMetadata, 352 ) -> Result<(), RepositoryOperationError> { 353 match read_latest_config_binding(transaction).await? { 354 Some(actual) if actual.matches_current_configuration(expected) => Ok(()), 355 Some(_) | None => Err(RepositoryOperationError::Binding), 356 } 357 } 358 359 pub(crate) async fn read_latest_config_binding( 360 transaction: &mut ServiceSqliteTransaction<'_>, 361 ) -> Result<Option<PersistedMetadata>, RepositoryOperationError> { 362 let rows = sqlx::query(READ_LATEST_CONFIG_BINDING_SQL) 363 .fetch_all(&mut *transaction) 364 .await 365 .map_err(|_| RepositoryOperationError::Storage)?; 366 if rows.len() > 1 { 367 return Err(RepositoryOperationError::Binding); 368 } 369 rows.first().map(decode_metadata_row).transpose() 370 } 371 372 async fn read_metadata( 373 transaction: &mut ServiceSqliteTransaction<'_>, 374 ) -> Result<Option<PersistedMetadata>, RepositoryOperationError> { 375 let rows = sqlx::query(READ_METADATA_SQL) 376 .fetch_all(&mut *transaction) 377 .await 378 .map_err(|_| RepositoryOperationError::Storage)?; 379 if rows.len() > 1 { 380 return Err(RepositoryOperationError::Binding); 381 } 382 let Some(row) = rows.first() else { 383 return Ok(None); 384 }; 385 let singleton = row 386 .try_get::<i64, _>("singleton") 387 .map_err(|_| RepositoryOperationError::Binding)?; 388 let actual = decode_metadata_row(row)?; 389 (singleton == 1) 390 .then_some(Some(actual)) 391 .ok_or(RepositoryOperationError::Binding) 392 } 393 394 fn decode_metadata_row( 395 row: &sqlx::sqlite::SqliteRow, 396 ) -> Result<PersistedMetadata, RepositoryOperationError> { 397 let normalized = row 398 .try_get::<Option<Vec<u8>>, _>("normalized_config_sha256") 399 .map_err(|_| RepositoryOperationError::Binding)? 400 .ok_or(RepositoryOperationError::Binding)?; 401 let normalized_config_sha256 = normalized 402 .try_into() 403 .map_err(|_| RepositoryOperationError::Binding)?; 404 let transport_public_key = bounded_public_key(row, "transport_public_key")?; 405 let user_public_key = bounded_public_key(row, "user_public_key")?; 406 let discovery_type = row 407 .try_get::<&str, _>("discovery_public_key_type") 408 .map_err(|_| RepositoryOperationError::Binding)?; 409 let discovery_public_key = match discovery_type { 410 "null" => None, 411 "text" => Some(bounded_public_key(row, "discovery_public_key")?), 412 _ => return Err(RepositoryOperationError::Binding), 413 }; 414 Ok(PersistedMetadata { 415 normalized_config_sha256, 416 transport_public_key, 417 user_public_key, 418 discovery_public_key, 419 config_contract_version: bounded_version(row, "config_contract_version")?, 420 state_contract_version: bounded_version(row, "state_contract_version")?, 421 operator_contract_version: bounded_version(row, "operator_contract_version")?, 422 status_contract_version: bounded_version(row, "status_contract_version")?, 423 }) 424 } 425 426 fn bounded_public_key( 427 row: &sqlx::sqlite::SqliteRow, 428 column: &str, 429 ) -> Result<Box<str>, RepositoryOperationError> { 430 let value = row 431 .try_get::<Option<String>, _>(column) 432 .map_err(|_| RepositoryOperationError::Binding)? 433 .ok_or(RepositoryOperationError::Binding)?; 434 let valid = value.len() == 64 435 && value.as_bytes().iter().all(u8::is_ascii_hexdigit) 436 && !value.as_bytes().iter().any(u8::is_ascii_uppercase); 437 valid 438 .then(|| value.into_boxed_str()) 439 .ok_or(RepositoryOperationError::Binding) 440 } 441 442 fn bounded_version( 443 row: &sqlx::sqlite::SqliteRow, 444 column: &str, 445 ) -> Result<u32, RepositoryOperationError> { 446 let value = row 447 .try_get::<i64, _>(column) 448 .map_err(|_| RepositoryOperationError::Binding)?; 449 u32::try_from(value) 450 .ok() 451 .filter(|value| *value != 0) 452 .ok_or(RepositoryOperationError::Binding) 453 } 454 455 async fn insert_metadata( 456 transaction: &mut ServiceSqliteTransaction<'_>, 457 expected: &PersistedMetadata, 458 ) -> Result<(), RepositoryOperationError> { 459 let result = sqlx::query(INSERT_METADATA_SQL) 460 .bind(expected.normalized_config_sha256.as_slice()) 461 .bind(expected.transport_public_key.as_ref()) 462 .bind(expected.user_public_key.as_ref()) 463 .bind(expected.discovery_public_key.as_deref()) 464 .bind(i64::from(expected.config_contract_version)) 465 .bind(i64::from(expected.state_contract_version)) 466 .bind(i64::from(expected.operator_contract_version)) 467 .bind(i64::from(expected.status_contract_version)) 468 .execute(&mut *transaction) 469 .await 470 .map_err(|_| RepositoryOperationError::Storage)?; 471 (result.rows_affected() == 1) 472 .then_some(()) 473 .ok_or(RepositoryOperationError::Storage) 474 } 475 476 async fn insert_initial_config_binding( 477 transaction: &mut ServiceSqliteTransaction<'_>, 478 ) -> Result<(), RepositoryOperationError> { 479 let result = sqlx::query(INSERT_INITIAL_CONFIG_BINDING_SQL) 480 .bind(i64::from(MYC_STATE_SCHEMA_VERSION)) 481 .bind(i64::from(MYC_STATE_SCHEMA_VERSION)) 482 .execute(&mut *transaction) 483 .await 484 .map_err(|_| RepositoryOperationError::Storage)?; 485 (result.rows_affected() == 1) 486 .then_some(()) 487 .ok_or(RepositoryOperationError::Storage) 488 } 489 490 fn map_transaction_error( 491 error: ServiceSqliteTransactionError<RepositoryOperationError>, 492 ) -> MycStateRepositoryError { 493 if error.kind() == ServiceSqliteTransactionErrorKind::CommitOutcomeUnknown { 494 return MycStateRepositoryError::new(MycStateRepositoryErrorKind::CommitOutcomeUnknown); 495 } 496 let kind = match error.operation_error() { 497 Some(RepositoryOperationError::Binding) => MycStateRepositoryErrorKind::Binding, 498 Some(RepositoryOperationError::Storage) | None => MycStateRepositoryErrorKind::Transaction, 499 }; 500 MycStateRepositoryError::new(kind) 501 }