migration.rs (37204B)
1 //! Governed SQLite schema migration boundary. 2 3 use crate::{Error, OpenMode}; 4 use sqlx::{Connection, Row, SqliteConnection, sqlite::SqliteConnectOptions}; 5 6 mod authored_v10; 7 pub use authored_v10::AuthoredV10Preflight; 8 9 /// Inspects an exact V10 runtime database without mutating it. 10 #[cfg_attr(coverage_nightly, coverage(off))] 11 pub async fn preflight_authored_v10(paths: &crate::Paths) -> Result<AuthoredV10Preflight, Error> { 12 paths.validate_filesystem(OpenMode::ReadOnly)?; 13 let mut connection = SqliteConnection::connect_with( 14 &SqliteConnectOptions::new() 15 .filename(paths.runtime()) 16 .read_only(true), 17 ) 18 .await 19 .map_err(|source| crate::open::map_database_open_error(&source, RUNTIME_DATABASE))?; 20 sqlx::raw_sql("PRAGMA query_only = ON") 21 .execute(&mut connection) 22 .await 23 .map_err(|source| { 24 crate::backend::startup_error( 25 &source, 26 Error::SchemaMetadataUnavailable { 27 database: RUNTIME_DATABASE, 28 }, 29 ) 30 })?; 31 let current = metadata(&mut connection, RUNTIME_DATABASE).await?; 32 if current.application_id != RUNTIME_APPLICATION_ID { 33 return Err(Error::SchemaIdentityMismatch { 34 database: RUNTIME_DATABASE, 35 expected: RUNTIME_APPLICATION_ID, 36 actual: current.application_id, 37 }); 38 } 39 if current.version < 10 { 40 return Err(Error::SchemaMigrationRequired { 41 database: RUNTIME_DATABASE, 42 current: 10, 43 actual: current.version, 44 }); 45 } 46 if current.version > 10 { 47 return Err(Error::SchemaTooNew { 48 database: RUNTIME_DATABASE, 49 supported: 10, 50 actual: current.version, 51 }); 52 } 53 validate_exact_catalog( 54 &mut connection, 55 RUNTIME_DATABASE, 56 10, 57 runtime::MIGRATIONS[9].owned_objects(), 58 ) 59 .await?; 60 let report = authored_v10::inspect(&mut connection).await?.report; 61 connection.close().await.map_err(|source| { 62 crate::backend::startup_error( 63 &source, 64 Error::DatabaseCloseFailed { 65 database: RUNTIME_DATABASE, 66 }, 67 ) 68 })?; 69 Ok(report) 70 } 71 72 /// Versioned schema authority for `private.sqlite`. 73 pub mod private; 74 /// Versioned schema authority for `runtime.sqlite`. 75 pub mod runtime; 76 77 const RUNTIME_DATABASE: &str = "runtime.sqlite"; 78 const PRIVATE_DATABASE: &str = "private.sqlite"; 79 const RUNTIME_APPLICATION_ID: u32 = 1_380_209_236; 80 const PRIVATE_APPLICATION_ID: u32 = 1_380_208_722; 81 82 const SET_RUNTIME_APPLICATION_ID: &str = "PRAGMA application_id = 1380209236"; 83 const SET_PRIVATE_APPLICATION_ID: &str = "PRAGMA application_id = 1380208722"; 84 85 #[derive(Clone, Copy)] 86 struct MigrationStep { 87 version: u32, 88 sql: &'static str, 89 owned_objects: &'static [&'static str], 90 } 91 92 struct MigrationPlan { 93 database: &'static str, 94 application_id: u32, 95 set_application_id_sql: &'static str, 96 minimum_version: u32, 97 current_version: u32, 98 steps: Vec<MigrationStep>, 99 } 100 101 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 102 pub(crate) struct MigrationReport { 103 initial_version: u32, 104 final_version: u32, 105 applied: u32, 106 } 107 108 impl MigrationReport { 109 #[allow(dead_code)] // Read by the public open lifecycle in its ordered RCL checkpoint. 110 pub(crate) const fn initial_version(self) -> u32 { 111 self.initial_version 112 } 113 114 #[allow(dead_code)] // Read by the public open lifecycle in its ordered RCL checkpoint. 115 pub(crate) const fn final_version(self) -> u32 { 116 self.final_version 117 } 118 119 #[allow(dead_code)] // Read by the public open lifecycle in its ordered RCL checkpoint. 120 pub(crate) const fn applied(self) -> u32 { 121 self.applied 122 } 123 } 124 125 #[cfg_attr(coverage_nightly, coverage(off))] 126 pub(crate) async fn migrate_runtime( 127 connection: &mut SqliteConnection, 128 mode: OpenMode, 129 ) -> Result<MigrationReport, Error> { 130 migrate(connection, mode, &runtime_plan()?).await 131 } 132 133 #[cfg_attr(coverage_nightly, coverage(off))] 134 pub(crate) async fn migrate_private( 135 connection: &mut SqliteConnection, 136 mode: OpenMode, 137 ) -> Result<MigrationReport, Error> { 138 migrate(connection, mode, &private_plan()?).await 139 } 140 141 fn runtime_plan() -> Result<MigrationPlan, Error> { 142 let steps = runtime::MIGRATIONS 143 .iter() 144 .map(|migration| { 145 Ok(MigrationStep { 146 version: migration.version(), 147 sql: runtime::migration_sql(migration.version()).ok_or( 148 Error::SchemaMetadataUnavailable { 149 database: RUNTIME_DATABASE, 150 }, 151 )?, 152 owned_objects: migration.owned_objects(), 153 }) 154 }) 155 .collect::<Result<Vec<_>, Error>>()?; 156 Ok(MigrationPlan { 157 database: RUNTIME_DATABASE, 158 application_id: RUNTIME_APPLICATION_ID, 159 set_application_id_sql: SET_RUNTIME_APPLICATION_ID, 160 minimum_version: runtime::MINIMUM_VERSION, 161 current_version: runtime::CURRENT_VERSION, 162 steps, 163 }) 164 } 165 166 fn private_plan() -> Result<MigrationPlan, Error> { 167 let steps = private::MIGRATIONS 168 .iter() 169 .map(|migration| { 170 Ok(MigrationStep { 171 version: migration.version(), 172 sql: private::migration_sql(migration.version()).ok_or( 173 Error::SchemaMetadataUnavailable { 174 database: PRIVATE_DATABASE, 175 }, 176 )?, 177 owned_objects: migration.owned_objects(), 178 }) 179 }) 180 .collect::<Result<Vec<_>, Error>>()?; 181 Ok(MigrationPlan { 182 database: PRIVATE_DATABASE, 183 application_id: PRIVATE_APPLICATION_ID, 184 set_application_id_sql: SET_PRIVATE_APPLICATION_ID, 185 minimum_version: private::MINIMUM_VERSION, 186 current_version: private::CURRENT_VERSION, 187 steps, 188 }) 189 } 190 191 // Validate both existing members before WAL setup or any forward migration. 192 // The caller retains the canonical writer lock; migration still rechecks under 193 // its own transaction. This is compatibility preflight, not a cross-file commit. 194 pub(crate) async fn preflight_existing(options: &crate::OpenOptions) -> Result<(), Error> { 195 for (path, plan) in [ 196 (options.paths().runtime(), runtime_plan()?), 197 (options.paths().private(), private_plan()?), 198 ] { 199 if !path.try_exists().map_err(|source| Error::Inspect { 200 path: path.to_path_buf(), 201 source, 202 })? { 203 if options.mode().may_create() { 204 continue; 205 } 206 return Err(Error::MissingFile(path.to_path_buf())); 207 } 208 let mut connection = SqliteConnection::connect_with( 209 &SqliteConnectOptions::new() 210 .filename(path) 211 .read_only(true) 212 .busy_timeout(options.busy_timeout()) 213 .pragma("query_only", "ON"), 214 ) 215 .await 216 .map_err(|source| crate::open::map_database_open_error(&source, plan.database))?; 217 let inspected = async { 218 let metadata = inspect(&mut connection, options.mode(), &plan).await?; 219 if plan.database == RUNTIME_DATABASE && metadata.version > 0 { 220 crate::open::preflight_source_generation( 221 &mut connection, 222 options.mode(), 223 options.source_generation_bootstrap(), 224 ) 225 .await?; 226 } 227 Ok::<(), Error>(()) 228 } 229 .await; 230 let closed = connection.close().await.map_err(|source| { 231 crate::backend::startup_error( 232 &source, 233 Error::DatabaseCloseFailed { 234 database: plan.database, 235 }, 236 ) 237 }); 238 inspected?; 239 closed?; 240 } 241 Ok(()) 242 } 243 244 async fn inspect( 245 connection: &mut SqliteConnection, 246 mode: OpenMode, 247 plan: &MigrationPlan, 248 ) -> Result<SchemaMetadata, Error> { 249 validate_plan(plan)?; 250 let initial = metadata(connection, plan.database).await?; 251 validate_metadata(plan, initial)?; 252 validate_catalog(connection, plan, initial.version).await?; 253 if plan.database == RUNTIME_DATABASE && initial.version == 10 { 254 let inspected = authored_v10::inspect(connection).await?; 255 if !inspected.report.is_eligible() { 256 return Err(inspected.report.blocked_error()); 257 } 258 } 259 // Private v4's pinned SQL refuses old v2 envelopes without a context 260 // fingerprint. Check the same eligibility before either file is migrated; 261 // keep the SQL guard as the transactional authority. 262 if mode.is_writable() && plan.database == PRIVATE_DATABASE && (1..4).contains(&initial.version) 263 { 264 let unsupported = sqlx::query_scalar::<_, bool>( 265 "SELECT EXISTS(SELECT 1 FROM radroots_private_artifacts WHERE envelope_version = 2)", 266 ) 267 .fetch_one(&mut *connection) 268 .await 269 .map_err(|source| schema_metadata_error(&source, PRIVATE_DATABASE))?; 270 if unsupported { 271 return Err(Error::SchemaMigrationFailed { 272 database: PRIVATE_DATABASE, 273 target_version: 4, 274 }); 275 } 276 } 277 if initial.version != plan.current_version && !mode.is_writable() { 278 return Err(Error::SchemaMigrationRequired { 279 database: plan.database, 280 current: plan.current_version, 281 actual: initial.version, 282 }); 283 } 284 Ok(initial) 285 } 286 287 #[cfg_attr(coverage_nightly, coverage(off))] 288 async fn migrate( 289 connection: &mut SqliteConnection, 290 mode: OpenMode, 291 plan: &MigrationPlan, 292 ) -> Result<MigrationReport, Error> { 293 let initial = inspect(connection, mode, plan).await?; 294 if initial.version == plan.current_version { 295 return Ok(MigrationReport { 296 initial_version: initial.version, 297 final_version: initial.version, 298 applied: 0, 299 }); 300 } 301 302 let mut transaction = connection 303 .begin_with("BEGIN IMMEDIATE") 304 .await 305 .map_err(|source| { 306 crate::backend::startup_error( 307 &source, 308 Error::SchemaMigrationFailed { 309 database: plan.database, 310 target_version: initial.version.saturating_add(1), 311 }, 312 ) 313 })?; 314 let transactional = match metadata(&mut transaction, plan.database).await { 315 Ok(metadata) => metadata, 316 Err(error) => { 317 let _rollback = transaction.rollback().await; 318 return Err(error); 319 } 320 }; 321 if transactional != initial { 322 let _rollback = transaction.rollback().await; 323 return Err(Error::SchemaMetadataUnavailable { 324 database: plan.database, 325 }); 326 } 327 328 if initial.version == 0 329 && let Err(source) = sqlx::raw_sql(plan.set_application_id_sql) 330 .execute(&mut *transaction) 331 .await 332 { 333 let error = crate::backend::startup_error( 334 &source, 335 Error::SchemaMigrationFailed { 336 database: plan.database, 337 target_version: 1, 338 }, 339 ); 340 let _rollback = transaction.rollback().await; 341 return Err(error); 342 } 343 344 let mut applied = 0_u32; 345 for step in plan 346 .steps 347 .iter() 348 .filter(|step| step.version > initial.version) 349 { 350 let inspected_v10 = if plan.database == RUNTIME_DATABASE && step.version == 11 { 351 match authored_v10::inspect(&mut transaction).await { 352 Ok(inspected) if inspected.report.is_eligible() => Some(inspected), 353 Ok(inspected) => { 354 let error = inspected.report.blocked_error(); 355 let _rollback = transaction.rollback().await; 356 return Err(error); 357 } 358 Err(error) => { 359 let _rollback = transaction.rollback().await; 360 return Err(error); 361 } 362 } 363 } else { 364 None 365 }; 366 let version_sql = 367 set_user_version_sql(step.version).ok_or(Error::SchemaMigrationFailed { 368 database: plan.database, 369 target_version: step.version, 370 })?; 371 if let Err(source) = sqlx::raw_sql(step.sql).execute(&mut *transaction).await { 372 let error = crate::backend::startup_error( 373 &source, 374 Error::SchemaMigrationFailed { 375 database: plan.database, 376 target_version: step.version, 377 }, 378 ); 379 let _rollback = transaction.rollback().await; 380 return Err(error); 381 } 382 if let Some(inspected) = inspected_v10.as_ref() 383 && let Err(error) = authored_v10::apply(&mut transaction, inspected).await 384 { 385 let _rollback = transaction.rollback().await; 386 return Err(error); 387 } 388 if let Err(source) = sqlx::raw_sql(version_sql).execute(&mut *transaction).await { 389 let error = crate::backend::startup_error( 390 &source, 391 Error::SchemaMigrationFailed { 392 database: plan.database, 393 target_version: step.version, 394 }, 395 ); 396 let _rollback = transaction.rollback().await; 397 return Err(error); 398 } 399 if let Err(source) = validate_exact_catalog( 400 &mut transaction, 401 plan.database, 402 step.version, 403 step.owned_objects, 404 ) 405 .await 406 { 407 let error = match source { 408 Error::SpaceInsufficient => Error::SpaceInsufficient, 409 _ => Error::SchemaMigrationFailed { 410 database: plan.database, 411 target_version: step.version, 412 }, 413 }; 414 let _rollback = transaction.rollback().await; 415 return Err(error); 416 } 417 applied = applied.saturating_add(1); 418 } 419 transaction.commit().await.map_err(|source| { 420 crate::backend::startup_error( 421 &source, 422 Error::SchemaMigrationFailed { 423 database: plan.database, 424 target_version: plan.current_version, 425 }, 426 ) 427 })?; 428 Ok(MigrationReport { 429 initial_version: initial.version, 430 final_version: plan.current_version, 431 applied, 432 }) 433 } 434 435 #[derive(Clone, Copy, Eq, PartialEq)] 436 struct SchemaMetadata { 437 application_id: u32, 438 version: u32, 439 } 440 441 #[cfg_attr(coverage_nightly, coverage(off))] 442 async fn metadata( 443 connection: &mut SqliteConnection, 444 database: &'static str, 445 ) -> Result<SchemaMetadata, Error> { 446 let application_id = sqlx::query_scalar::<_, i64>("PRAGMA application_id") 447 .fetch_one(&mut *connection) 448 .await 449 .map_err(|source| schema_metadata_error(&source, database))?; 450 let version = sqlx::query_scalar::<_, i64>("PRAGMA user_version") 451 .fetch_one(&mut *connection) 452 .await 453 .map_err(|source| schema_metadata_error(&source, database))?; 454 Ok(SchemaMetadata { 455 application_id: u32::try_from(application_id) 456 .map_err(|_| Error::SchemaMetadataUnavailable { database })?, 457 version: u32::try_from(version) 458 .map_err(|_| Error::SchemaMetadataUnavailable { database })?, 459 }) 460 } 461 462 fn schema_metadata_error(source: &sqlx::Error, database: &'static str) -> Error { 463 match crate::open::map_database_open_error(source, database) { 464 error @ (Error::DatabaseCorrupt { .. } | Error::SpaceInsufficient) => error, 465 _ => Error::SchemaMetadataUnavailable { database }, 466 } 467 } 468 469 fn validate_plan(plan: &MigrationPlan) -> Result<(), Error> { 470 let valid = plan.minimum_version > 0 471 && plan.minimum_version <= plan.current_version 472 && plan.current_version <= 17 473 && plan.steps.len() == usize::try_from(plan.current_version).unwrap_or(usize::MAX) 474 && plan 475 .steps 476 .iter() 477 .enumerate() 478 .all(|(index, step)| step.version == u32::try_from(index + 1).unwrap_or(u32::MAX)); 479 if valid { 480 Ok(()) 481 } else { 482 Err(Error::SchemaMetadataUnavailable { 483 database: plan.database, 484 }) 485 } 486 } 487 488 fn validate_metadata(plan: &MigrationPlan, metadata: SchemaMetadata) -> Result<(), Error> { 489 if metadata.version > plan.current_version { 490 return Err(Error::SchemaTooNew { 491 database: plan.database, 492 supported: plan.current_version, 493 actual: metadata.version, 494 }); 495 } 496 if metadata.version > 0 && metadata.version < plan.minimum_version { 497 return Err(Error::SchemaTooOld { 498 database: plan.database, 499 minimum: plan.minimum_version, 500 actual: metadata.version, 501 }); 502 } 503 let expected_application_id = if metadata.version == 0 { 504 0 505 } else { 506 plan.application_id 507 }; 508 if metadata.application_id != expected_application_id { 509 return Err(Error::SchemaIdentityMismatch { 510 database: plan.database, 511 expected: expected_application_id, 512 actual: metadata.application_id, 513 }); 514 } 515 Ok(()) 516 } 517 518 #[cfg_attr(coverage_nightly, coverage(off))] 519 async fn validate_catalog( 520 connection: &mut SqliteConnection, 521 plan: &MigrationPlan, 522 version: u32, 523 ) -> Result<(), Error> { 524 let expected = if version == 0 { 525 &[][..] 526 } else { 527 plan.steps 528 .get( 529 usize::try_from(version - 1).map_err(|_| Error::SchemaCatalogMismatch { 530 database: plan.database, 531 version, 532 })?, 533 ) 534 .ok_or(Error::SchemaCatalogMismatch { 535 database: plan.database, 536 version, 537 })? 538 .owned_objects 539 }; 540 validate_exact_catalog(connection, plan.database, version, expected).await 541 } 542 543 #[cfg_attr(coverage_nightly, coverage(off))] 544 async fn validate_exact_catalog( 545 connection: &mut SqliteConnection, 546 database: &'static str, 547 version: u32, 548 expected: &[&str], 549 ) -> Result<(), Error> { 550 let rows = sqlx::query( 551 "SELECT name FROM sqlite_schema 552 WHERE name NOT LIKE 'sqlite_%' 553 ORDER BY name", 554 ) 555 .fetch_all(&mut *connection) 556 .await 557 .map_err(|source| { 558 crate::backend::startup_error(&source, Error::SchemaCatalogMismatch { database, version }) 559 })?; 560 let actual = rows 561 .iter() 562 .map(|row| row.get::<String, _>("name")) 563 .collect::<Vec<_>>(); 564 if actual 565 .iter() 566 .map(String::as_str) 567 .eq(expected.iter().copied()) 568 { 569 Ok(()) 570 } else if version == 0 { 571 Err(Error::UnrecognizedSchema { database }) 572 } else { 573 Err(Error::SchemaCatalogMismatch { database, version }) 574 } 575 } 576 577 const fn set_user_version_sql(version: u32) -> Option<&'static str> { 578 match version { 579 1 => Some("PRAGMA user_version = 1"), 580 2 => Some("PRAGMA user_version = 2"), 581 3 => Some("PRAGMA user_version = 3"), 582 4 => Some("PRAGMA user_version = 4"), 583 5 => Some("PRAGMA user_version = 5"), 584 6 => Some("PRAGMA user_version = 6"), 585 7 => Some("PRAGMA user_version = 7"), 586 8 => Some("PRAGMA user_version = 8"), 587 9 => Some("PRAGMA user_version = 9"), 588 10 => Some("PRAGMA user_version = 10"), 589 11 => Some("PRAGMA user_version = 11"), 590 12 => Some("PRAGMA user_version = 12"), 591 13 => Some("PRAGMA user_version = 13"), 592 14 => Some("PRAGMA user_version = 14"), 593 15 => Some("PRAGMA user_version = 15"), 594 16 => Some("PRAGMA user_version = 16"), 595 17 => Some("PRAGMA user_version = 17"), 596 _ => None, 597 } 598 } 599 600 #[cfg(test)] 601 #[cfg_attr(coverage_nightly, coverage(off))] 602 mod tests { 603 use sqlx::sqlite::SqliteConnectOptions; 604 605 use super::*; 606 607 const TEST_V1_OBJECTS: &[&str] = &["radroots_test_one"]; 608 const TEST_V2_OBJECTS: &[&str] = &["radroots_test_one", "radroots_test_two"]; 609 610 pub(super) async fn connection() -> SqliteConnection { 611 SqliteConnection::connect("sqlite::memory:") 612 .await 613 .expect("memory SQLite") 614 } 615 616 pub(super) async fn pragma(connection: &mut SqliteConnection, name: &str) -> i64 { 617 let sql = match name { 618 "application_id" => "PRAGMA application_id", 619 "user_version" => "PRAGMA user_version", 620 _ => panic!("unsupported test pragma"), 621 }; 622 sqlx::query_scalar(sql) 623 .fetch_one(connection) 624 .await 625 .expect("pragma") 626 } 627 628 pub(super) async fn establish_runtime_version(connection: &mut SqliteConnection, version: u32) { 629 for migration_version in 1..=version { 630 sqlx::raw_sql( 631 runtime::migration_sql(migration_version).expect("registered runtime SQL"), 632 ) 633 .execute(&mut *connection) 634 .await 635 .expect("runtime migration"); 636 } 637 sqlx::raw_sql(SET_RUNTIME_APPLICATION_ID) 638 .execute(&mut *connection) 639 .await 640 .expect("runtime application id"); 641 sqlx::raw_sql(set_user_version_sql(version).expect("version pragma")) 642 .execute(&mut *connection) 643 .await 644 .expect("runtime user version"); 645 } 646 647 pub(super) async fn establish_private_version(connection: &mut SqliteConnection, version: u32) { 648 for migration_version in 1..=version { 649 sqlx::raw_sql( 650 private::migration_sql(migration_version).expect("registered private SQL"), 651 ) 652 .execute(&mut *connection) 653 .await 654 .expect("private migration"); 655 } 656 sqlx::raw_sql(SET_PRIVATE_APPLICATION_ID) 657 .execute(&mut *connection) 658 .await 659 .expect("private application id"); 660 sqlx::raw_sql(set_user_version_sql(version).expect("version pragma")) 661 .execute(&mut *connection) 662 .await 663 .expect("private user version"); 664 } 665 666 #[tokio::test] 667 async fn fresh_runtime_and_private_schemas_migrate_to_exact_current_versions() { 668 let mut runtime_connection = connection().await; 669 let runtime_report = migrate_runtime(&mut runtime_connection, OpenMode::Create) 670 .await 671 .expect("runtime migrations"); 672 assert_eq!(runtime_report.initial_version(), 0); 673 assert_eq!(runtime_report.final_version(), runtime::CURRENT_VERSION); 674 assert_eq!(runtime_report.applied(), runtime::CURRENT_VERSION); 675 assert_eq!( 676 pragma(&mut runtime_connection, "application_id").await, 677 i64::from(RUNTIME_APPLICATION_ID) 678 ); 679 assert_eq!( 680 pragma(&mut runtime_connection, "user_version").await, 681 i64::from(runtime::CURRENT_VERSION) 682 ); 683 assert_eq!( 684 migrate_runtime(&mut runtime_connection, OpenMode::ReadOnly) 685 .await 686 .expect("current read-only runtime"), 687 MigrationReport { 688 initial_version: runtime::CURRENT_VERSION, 689 final_version: runtime::CURRENT_VERSION, 690 applied: 0, 691 } 692 ); 693 694 let mut private_connection = connection().await; 695 let private_report = migrate_private(&mut private_connection, OpenMode::Create) 696 .await 697 .expect("private migrations"); 698 assert_eq!(private_report.initial_version(), 0); 699 assert_eq!(private_report.final_version(), private::CURRENT_VERSION); 700 assert_eq!(private_report.applied(), private::CURRENT_VERSION); 701 assert_eq!( 702 pragma(&mut private_connection, "application_id").await, 703 i64::from(PRIVATE_APPLICATION_ID) 704 ); 705 assert_eq!( 706 pragma(&mut private_connection, "user_version").await, 707 i64::from(private::CURRENT_VERSION) 708 ); 709 } 710 711 #[tokio::test] 712 async fn recognized_runtime_schema_upgrades_forward_and_preserves_data() { 713 let mut connection = connection().await; 714 establish_runtime_version(&mut connection, 1).await; 715 sqlx::query( 716 "INSERT INTO radroots_runtime_source_generations ( 717 generation, state, created_at_unix_ms 718 ) VALUES (?, 'active', 10)", 719 ) 720 .bind([7_u8; 32].as_slice()) 721 .execute(&mut connection) 722 .await 723 .expect("v1 data"); 724 725 let report = migrate_runtime(&mut connection, OpenMode::ReadWriteExisting) 726 .await 727 .expect("forward migration"); 728 assert_eq!(report.initial_version(), 1); 729 assert_eq!(report.final_version(), runtime::CURRENT_VERSION); 730 assert_eq!(report.applied(), runtime::CURRENT_VERSION - 1); 731 assert_eq!( 732 sqlx::query_scalar::<_, i64>( 733 "SELECT COUNT(*) FROM radroots_runtime_source_generations", 734 ) 735 .fetch_one(&mut connection) 736 .await 737 .expect("preserved data"), 738 1 739 ); 740 } 741 742 #[tokio::test] 743 async fn capacity_migration_retains_original_generation_and_retries_exact_pending_suffix() { 744 let mut connection = connection().await; 745 establish_runtime_version(&mut connection, 1).await; 746 sqlx::query("INSERT INTO radroots_runtime_source_generations (generation, state, created_at_unix_ms) VALUES (?, 'active', 10)") 747 .bind([7_u8; 32].as_slice()).execute(&mut connection).await.unwrap(); 748 let page_count: i64 = sqlx::query_scalar("PRAGMA page_count") 749 .fetch_one(&mut connection) 750 .await 751 .unwrap(); 752 let previous_limit: i64 = sqlx::query_scalar("PRAGMA max_page_count") 753 .fetch_one(&mut connection) 754 .await 755 .unwrap(); 756 // PRAGMA assignments cannot bind; the interpolated value is an i64 from SQLite. 757 sqlx::raw_sql(sqlx::AssertSqlSafe(format!( 758 "PRAGMA max_page_count = {page_count}" 759 ))) 760 .execute(&mut connection) 761 .await 762 .unwrap(); 763 assert!(matches!( 764 migrate_runtime(&mut connection, OpenMode::ReadWriteExisting).await, 765 Err(Error::SpaceInsufficient) 766 )); 767 assert_eq!(pragma(&mut connection, "user_version").await, 1); 768 let generation: Vec<u8> = sqlx::query_scalar( 769 "SELECT generation FROM radroots_runtime_source_generations WHERE state = 'active'", 770 ) 771 .fetch_one(&mut connection) 772 .await 773 .unwrap(); 774 assert_eq!(generation, [7_u8; 32]); 775 // The restored limit is likewise an i64, never untrusted SQL text. 776 sqlx::raw_sql(sqlx::AssertSqlSafe(format!( 777 "PRAGMA max_page_count = {previous_limit}" 778 ))) 779 .execute(&mut connection) 780 .await 781 .unwrap(); 782 let report = migrate_runtime(&mut connection, OpenMode::ReadWriteExisting) 783 .await 784 .unwrap(); 785 assert_eq!(report.initial_version(), 1); 786 assert_eq!(report.final_version(), runtime::CURRENT_VERSION); 787 assert_eq!(report.applied(), runtime::CURRENT_VERSION - 1); 788 assert_eq!( 789 sqlx::query_scalar::<_, Vec<u8>>( 790 "SELECT generation FROM radroots_runtime_source_generations WHERE state = 'active'" 791 ) 792 .fetch_one(&mut connection) 793 .await 794 .unwrap(), 795 generation 796 ); 797 } 798 799 #[tokio::test] 800 async fn committed_migration_reopens_as_current_after_a_lost_success_response() { 801 let directory = tempfile::tempdir().expect("database directory"); 802 let path = directory.path().join(RUNTIME_DATABASE); 803 let mut connection = SqliteConnection::connect_with( 804 &SqliteConnectOptions::new() 805 .filename(&path) 806 .create_if_missing(true), 807 ) 808 .await 809 .expect("open migration database"); 810 establish_runtime_version(&mut connection, 1).await; 811 let _lost_response = migrate_runtime(&mut connection, OpenMode::ReadWriteExisting) 812 .await 813 .expect("commit pending migration"); 814 connection.close().await.expect("simulate process exit"); 815 816 let mut reopened = SqliteConnection::connect_with( 817 &SqliteConnectOptions::new().filename(&path).read_only(true), 818 ) 819 .await 820 .expect("reopen migrated database"); 821 let report = migrate_runtime(&mut reopened, OpenMode::ReadOnly) 822 .await 823 .expect("recognize committed migration"); 824 assert_eq!(report.initial_version(), runtime::CURRENT_VERSION); 825 assert_eq!(report.final_version(), runtime::CURRENT_VERSION); 826 assert_eq!(report.applied(), 0); 827 } 828 829 #[tokio::test] 830 async fn every_recognized_runtime_version_applies_exactly_the_pending_suffix() { 831 for initial_version in 1..=runtime::CURRENT_VERSION { 832 let mut connection = connection().await; 833 establish_runtime_version(&mut connection, initial_version).await; 834 let report = migrate_runtime(&mut connection, OpenMode::ReadWriteExisting) 835 .await 836 .expect("recognized forward migration"); 837 assert_eq!(report.initial_version(), initial_version); 838 assert_eq!(report.final_version(), runtime::CURRENT_VERSION); 839 assert_eq!(report.applied(), runtime::CURRENT_VERSION - initial_version); 840 } 841 } 842 843 #[tokio::test] 844 async fn every_recognized_private_version_applies_exactly_the_pending_suffix() { 845 for initial_version in 1..=private::CURRENT_VERSION { 846 let mut connection = connection().await; 847 establish_private_version(&mut connection, initial_version).await; 848 let report = migrate_private(&mut connection, OpenMode::ReadWriteExisting) 849 .await 850 .expect("recognized private forward migration"); 851 assert_eq!(report.initial_version(), initial_version); 852 assert_eq!(report.final_version(), private::CURRENT_VERSION); 853 assert_eq!(report.applied(), private::CURRENT_VERSION - initial_version); 854 } 855 } 856 857 #[tokio::test] 858 async fn read_only_old_schema_requires_migration_without_mutating() { 859 let mut connection = connection().await; 860 establish_runtime_version(&mut connection, 1).await; 861 assert!(matches!( 862 migrate_runtime(&mut connection, OpenMode::ReadOnly).await, 863 Err(Error::SchemaMigrationRequired { 864 database: RUNTIME_DATABASE, 865 current: runtime::CURRENT_VERSION, 866 actual: 1, 867 }) 868 )); 869 assert_eq!(pragma(&mut connection, "user_version").await, 1); 870 } 871 872 #[tokio::test] 873 async fn newer_wrong_identity_and_unversioned_nonempty_schemas_fail_closed() { 874 let mut newer = connection().await; 875 sqlx::raw_sql(SET_RUNTIME_APPLICATION_ID) 876 .execute(&mut newer) 877 .await 878 .expect("application id"); 879 let newer_version = 18; 880 assert_eq!(newer_version, runtime::CURRENT_VERSION + 1); 881 sqlx::raw_sql("PRAGMA user_version = 18") 882 .execute(&mut newer) 883 .await 884 .expect("newer version"); 885 assert!(matches!( 886 migrate_runtime(&mut newer, OpenMode::ReadWriteExisting).await, 887 Err(Error::SchemaTooNew { 888 database: RUNTIME_DATABASE, 889 supported: runtime::CURRENT_VERSION, 890 actual, 891 }) if actual == newer_version 892 )); 893 assert_eq!( 894 pragma(&mut newer, "user_version").await, 895 i64::from(newer_version) 896 ); 897 898 let mut wrong_identity = connection().await; 899 establish_runtime_version(&mut wrong_identity, 1).await; 900 assert!(matches!( 901 migrate_private(&mut wrong_identity, OpenMode::ReadWriteExisting).await, 902 Err(Error::SchemaIdentityMismatch { 903 database: PRIVATE_DATABASE, 904 expected: PRIVATE_APPLICATION_ID, 905 actual: RUNTIME_APPLICATION_ID, 906 }) 907 )); 908 909 let mut unknown = connection().await; 910 sqlx::query("CREATE TABLE unrelated(value INTEGER)") 911 .execute(&mut unknown) 912 .await 913 .expect("unknown table"); 914 assert!(matches!( 915 migrate_runtime(&mut unknown, OpenMode::Create).await, 916 Err(Error::UnrecognizedSchema { 917 database: RUNTIME_DATABASE, 918 }) 919 )); 920 assert_eq!(pragma(&mut unknown, "application_id").await, 0); 921 assert_eq!(pragma(&mut unknown, "user_version").await, 0); 922 } 923 924 #[tokio::test] 925 async fn current_version_with_incomplete_catalog_is_rejected() { 926 let mut connection = connection().await; 927 sqlx::raw_sql(SET_PRIVATE_APPLICATION_ID) 928 .execute(&mut connection) 929 .await 930 .expect("private application id"); 931 sqlx::raw_sql("PRAGMA user_version = 1") 932 .execute(&mut connection) 933 .await 934 .expect("private version"); 935 assert!(matches!( 936 migrate_private(&mut connection, OpenMode::ReadOnly).await, 937 Err(Error::SchemaCatalogMismatch { 938 database: PRIVATE_DATABASE, 939 version: 1, 940 }) 941 )); 942 } 943 944 #[tokio::test] 945 async fn any_failed_step_rolls_back_the_entire_pending_plan() { 946 let mut connection = connection().await; 947 let plan = MigrationPlan { 948 database: "test.sqlite", 949 application_id: 4_242, 950 set_application_id_sql: "PRAGMA application_id = 4242", 951 minimum_version: 1, 952 current_version: 2, 953 steps: vec![ 954 MigrationStep { 955 version: 1, 956 sql: "CREATE TABLE radroots_test_one(value INTEGER)", 957 owned_objects: TEST_V1_OBJECTS, 958 }, 959 MigrationStep { 960 version: 2, 961 sql: "CREATE TABLE radroots_test_two(value INTEGER); 962 INSERT INTO radroots_missing VALUES (1)", 963 owned_objects: TEST_V2_OBJECTS, 964 }, 965 ], 966 }; 967 assert!(matches!( 968 migrate(&mut connection, OpenMode::Create, &plan).await, 969 Err(Error::SchemaMigrationFailed { 970 database: "test.sqlite", 971 target_version: 2, 972 }) 973 )); 974 assert_eq!(pragma(&mut connection, "application_id").await, 0); 975 assert_eq!(pragma(&mut connection, "user_version").await, 0); 976 assert_eq!( 977 sqlx::query_scalar::<_, i64>( 978 "SELECT COUNT(*) FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'", 979 ) 980 .fetch_one(&mut connection) 981 .await 982 .expect("rolled-back catalog"), 983 0 984 ); 985 } 986 987 #[test] 988 fn migration_plan_validation_rejects_every_invalid_shape() { 989 fn plan(minimum_version: u32, current_version: u32, versions: &[u32]) -> MigrationPlan { 990 MigrationPlan { 991 database: "test.sqlite", 992 application_id: 4_242, 993 set_application_id_sql: "PRAGMA application_id = 4242", 994 minimum_version, 995 current_version, 996 steps: versions 997 .iter() 998 .copied() 999 .map(|version| MigrationStep { 1000 version, 1001 sql: "SELECT 1", 1002 owned_objects: &[], 1003 }) 1004 .collect(), 1005 } 1006 } 1007 1008 assert!(validate_plan(&plan(1, 2, &[1, 2])).is_ok()); 1009 for invalid in [ 1010 plan(0, 2, &[1, 2]), 1011 plan(3, 2, &[1, 2]), 1012 plan(1, 10, &[1, 2]), 1013 plan(1, 2, &[1]), 1014 plan(1, 2, &[1, 3]), 1015 ] { 1016 assert!(matches!( 1017 validate_plan(&invalid), 1018 Err(Error::SchemaMetadataUnavailable { 1019 database: "test.sqlite" 1020 }) 1021 )); 1022 } 1023 } 1024 } 1025 1026 #[cfg(test)] 1027 #[cfg_attr(coverage_nightly, coverage(off))] 1028 #[path = "migration_draft_query_tests.rs"] 1029 mod draft_query_tests; 1030 1031 #[cfg(test)] 1032 #[cfg_attr(coverage_nightly, coverage(off))] 1033 #[path = "migration_signed_facts_tests.rs"] 1034 mod signed_facts_tests; 1035 1036 #[cfg(test)] 1037 #[cfg_attr(coverage_nightly, coverage(off))] 1038 #[path = "migration_delivery_facts_tests.rs"] 1039 mod delivery_facts_tests; 1040 1041 #[cfg(test)] 1042 #[cfg_attr(coverage_nightly, coverage(off))] 1043 #[path = "migration_delivery_reconciliation_tests.rs"] 1044 mod delivery_reconciliation_tests; 1045 1046 #[cfg(test)] 1047 #[path = "migration_preflight_tests.rs"] 1048 mod preflight_tests;