1use rusqlite::{Connection, OptionalExtension, Transaction};
59use std::collections::BTreeMap;
60use std::sync::{Mutex, OnceLock};
61
62use crate::error::{BridgeEligibility, SqliteStoreError};
63
64const CREATE_LEDGER_SQL: &str = "CREATE TABLE IF NOT EXISTS main.meerkat_schema (
65 domain TEXT PRIMARY KEY,
66 version INTEGER NOT NULL
67)";
68
69const CUSTODY_SAVEPOINT_SQL: &str = "SAVEPOINT meerkat_migration_custody";
74const CUSTODY_RELEASE_SQL: &str = "RELEASE SAVEPOINT meerkat_migration_custody";
75
76#[derive(Debug)]
78pub struct Migration {
79 pub version: i64,
82 pub name: &'static str,
84 pub apply: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
89}
90
91#[derive(Debug)]
93pub struct SchemaPredecessor {
94 pub version: i64,
95 pub verify: fn(&Connection) -> Result<(), String>,
96}
97
98#[derive(Debug, Clone, Copy, PartialEq, Eq)]
100pub enum SchemaObjectKind {
101 Table,
102 Index,
103 Trigger,
104 View,
105}
106
107impl SchemaObjectKind {
108 fn sqlite_name(self) -> &'static str {
109 match self {
110 Self::Table => "table",
111 Self::Index => "index",
112 Self::Trigger => "trigger",
113 Self::View => "view",
114 }
115 }
116}
117
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub struct SchemaObject {
126 pub kind: SchemaObjectKind,
127 pub name: &'static str,
128}
129
130#[derive(Debug)]
132pub struct SchemaDomain {
133 pub name: &'static str,
135 pub migrations: &'static [Migration],
137 pub initialize_current: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
145 pub allowed_existing_versions: &'static [i64],
149 pub released_predecessors: &'static [SchemaPredecessor],
151 pub bridge_recoverable_versions: &'static [i64],
163 pub owned_objects: &'static [SchemaObject],
166 pub retired_objects: &'static [SchemaObject],
170}
171
172impl SchemaDomain {
173 pub fn supported_version(&self) -> i64 {
175 self.migrations.last().map_or(0, |m| m.version)
176 }
177
178 fn validate(&self) -> Result<(), SqliteStoreError> {
179 for (idx, migration) in self.migrations.iter().enumerate() {
180 let expected = idx as i64 + 1;
181 if migration.version != expected {
182 return Err(SqliteStoreError::InvalidMigrationList {
183 domain: self.name.to_string(),
184 detail: format!(
185 "migration at position {idx} has version {}, expected {expected} \
186 (versions must be contiguous from 1)",
187 migration.version
188 ),
189 });
190 }
191 }
192 let supported = self.supported_version();
193 let mut previous = None;
194 for &version in self.allowed_existing_versions {
195 if version <= 0 || version > supported {
196 return Err(SqliteStoreError::InvalidMigrationList {
197 domain: self.name.to_string(),
198 detail: format!(
199 "allowed existing version {version} is outside 1..={supported}"
200 ),
201 });
202 }
203 if previous.is_some_and(|value| value >= version) {
204 return Err(SqliteStoreError::InvalidMigrationList {
205 domain: self.name.to_string(),
206 detail: "allowed existing versions must be strictly increasing".to_string(),
207 });
208 }
209 previous = Some(version);
210 }
211 if !self.allowed_existing_versions.contains(&supported) {
212 return Err(SqliteStoreError::InvalidMigrationList {
213 domain: self.name.to_string(),
214 detail: format!(
215 "allowed existing versions must explicitly include current version {supported}"
216 ),
217 });
218 }
219 for &version in self
220 .allowed_existing_versions
221 .iter()
222 .filter(|&&version| version < supported)
223 {
224 let matches = self
225 .released_predecessors
226 .iter()
227 .filter(|predecessor| predecessor.version == version)
228 .count();
229 if matches != 1 {
230 return Err(SqliteStoreError::InvalidMigrationList {
231 domain: self.name.to_string(),
232 detail: format!(
233 "allowed predecessor version {version} must have exactly one frozen \
234 verifier, found {matches}"
235 ),
236 });
237 }
238 }
239 validate_recoverable_source_versions(self, supported, self.bridge_recoverable_versions)?;
250 for predecessor in self.released_predecessors {
251 if predecessor.version >= supported
252 || !self
253 .allowed_existing_versions
254 .contains(&predecessor.version)
255 {
256 return Err(SqliteStoreError::InvalidMigrationList {
257 domain: self.name.to_string(),
258 detail: format!(
259 "fingerprint verifier for version {} is not an allowed predecessor",
260 predecessor.version
261 ),
262 });
263 }
264 }
265 for (idx, object) in self
266 .owned_objects
267 .iter()
268 .chain(self.retired_objects)
269 .enumerate()
270 {
271 if object.name.is_empty() || object.name == "meerkat_schema" {
272 return Err(SqliteStoreError::InvalidMigrationList {
273 domain: self.name.to_string(),
274 detail: format!(
275 "owned object at position {idx} has reserved or empty name `{}`",
276 object.name
277 ),
278 });
279 }
280 if self
281 .owned_objects
282 .iter()
283 .chain(self.retired_objects)
284 .take(idx)
285 .any(|prior| prior.name == object.name)
286 {
287 return Err(SqliteStoreError::InvalidMigrationList {
288 domain: self.name.to_string(),
289 detail: format!("owned object name `{}` is duplicated", object.name),
290 });
291 }
292 }
293 Ok(())
294 }
295
296 fn accepts_existing_version(&self, version: i64) -> bool {
297 self.allowed_existing_versions.contains(&version)
298 }
299
300 pub fn bridgeable_source_version(&self, conn: &Connection) -> Option<i64> {
324 if self.validate().is_err() {
325 return None;
326 }
327 authenticate_unledgered_source_version(
328 conn,
329 self,
330 self.supported_version(),
331 self.bridge_recoverable_versions,
332 Vec::new,
333 )
334 .ok()
335 .map(|matched| matched.version)
336 }
337
338 pub fn bridge_eligibility(&self, conn: &Connection) -> BridgeEligibility {
343 match self.bridgeable_source_version(conn) {
344 Some(_) => BridgeEligibility::CatalogAuthenticated,
345 None => BridgeEligibility::Unrecognized,
346 }
347 }
348
349 fn verify_stamped_version(
355 &self,
356 conn: &Connection,
357 version: i64,
358 ) -> Result<(), SqliteStoreError> {
359 if version != self.supported_version() {
360 return self.verify_predecessor(conn, version);
361 }
362 verify_current_schema_fingerprint(conn, self).map_err(|detail| {
363 match current_catalog_diff(conn, self) {
364 Ok(diff) if !diff.is_empty() => SqliteStoreError::CurrentSchemaMismatch {
365 domain: self.name.to_string(),
366 version,
367 missing_objects: diff.missing,
368 unexpected_objects: diff.unexpected,
369 changed_objects: diff.changed,
370 },
371 _ => SqliteStoreError::SchemaFingerprintMismatch {
372 domain: self.name.to_string(),
373 version,
374 detail,
375 },
376 }
377 })
378 }
379
380 fn verify_predecessor(&self, conn: &Connection, version: i64) -> Result<(), SqliteStoreError> {
381 if version == self.supported_version() {
382 return verify_current_schema_fingerprint(conn, self).map_err(|detail| {
383 SqliteStoreError::SchemaFingerprintMismatch {
384 domain: self.name.to_string(),
385 version,
386 detail,
387 }
388 });
389 }
390 let predecessor = self
391 .released_predecessors
392 .iter()
393 .find(|predecessor| predecessor.version == version)
394 .ok_or_else(|| unsupported_predecessor(self, version))?;
395 (predecessor.verify)(conn).map_err(|detail| SqliteStoreError::SchemaFingerprintMismatch {
396 domain: self.name.to_string(),
397 version,
398 detail,
399 })
400 }
401}
402
403#[derive(Debug, Clone, Copy, PartialEq, Eq)]
405pub struct LedgerReport {
406 pub from_version: i64,
408 pub to_version: i64,
410}
411
412impl LedgerReport {
413 pub fn migrated(&self) -> bool {
415 self.to_version > self.from_version
416 }
417}
418
419#[derive(Debug, Clone, PartialEq, Eq)]
429pub struct MaintenanceRecordRefusal {
430 pub record: String,
432 pub reason: String,
434}
435
436#[derive(Debug, Clone, Default, PartialEq, Eq)]
438pub struct MaintenancePrepareReport {
439 pub changed: usize,
441 pub refused: Vec<MaintenanceRecordRefusal>,
445}
446
447impl MaintenancePrepareReport {
448 pub fn rewrote(changed: usize) -> Self {
450 Self {
451 changed,
452 refused: Vec::new(),
453 }
454 }
455}
456
457#[derive(Debug, Clone, PartialEq, Eq)]
459pub struct MaintenanceBridgeReport {
460 pub from_version: i64,
462 pub to_version: i64,
464 pub prepared: usize,
466 pub refused: Vec<MaintenanceRecordRefusal>,
470}
471
472impl MaintenanceBridgeReport {
473 pub fn migrated(&self) -> bool {
475 self.to_version > self.from_version
476 }
477
478 pub fn changed(&self) -> bool {
480 self.migrated() || self.prepared > 0
481 }
482}
483
484pub fn domain_version(conn: &Connection, domain: &str) -> Result<Option<i64>, SqliteStoreError> {
492 if !ledger_table_exists(conn)? {
493 return Ok(None);
494 }
495 validate_ledger_shape(conn)?;
496 read_version(conn, domain)
497}
498
499pub fn preflight_schema_eligibility(
514 conn: &Connection,
515 domain: &SchemaDomain,
516) -> Result<(), SqliteStoreError> {
517 domain.validate()?;
518 let supported = domain.supported_version();
519 match domain_version(conn, domain.name)? {
520 Some(found) if found > supported => {
521 return Err(SqliteStoreError::SchemaFromTheFuture {
522 domain: domain.name.to_string(),
523 found,
524 supported,
525 });
526 }
527 Some(found) if !domain.accepts_existing_version(found) => {
528 return Err(unsupported_predecessor(domain, found));
529 }
530 Some(found) => domain.verify_stamped_version(conn, found)?,
531 None => {
532 let objects = find_owned_objects(conn, domain)?;
533 if !objects.is_empty() {
534 return Err(SqliteStoreError::UnledgeredDomainObjects {
535 domain: domain.name.to_string(),
536 objects,
537 bridgeable: domain.bridge_eligibility(conn),
538 });
539 }
540 }
541 }
542 Ok(())
543}
544
545pub fn apply_domain_migrations(
553 conn: &mut Connection,
554 domain: &SchemaDomain,
555) -> Result<LedgerReport, SqliteStoreError> {
556 domain.validate()?;
557 let supported = domain.supported_version();
558
559 {
560 let read = conn.transaction()?;
561 let is_current = domain_version(&read, domain.name)? == Some(supported);
562 if is_current {
563 domain.verify_stamped_version(&read, supported)?;
564 }
565 read.rollback()?;
566 if is_current {
567 return Ok(LedgerReport {
568 from_version: supported,
569 to_version: supported,
570 });
571 }
572 }
573
574 let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
575 let current = if ledger_table_exists(&tx)? {
578 validate_ledger_shape(&tx)?;
579 read_version(&tx, domain.name)?
580 } else {
581 None
582 };
583 if let Some(found) = current {
584 if found > supported {
585 return Err(SqliteStoreError::SchemaFromTheFuture {
586 domain: domain.name.to_string(),
587 found,
588 supported,
589 });
590 }
591 if !domain.accepts_existing_version(found) {
592 return Err(unsupported_predecessor(domain, found));
593 }
594 domain.verify_stamped_version(&tx, found)?;
595 } else {
596 let objects = find_owned_objects(&tx, domain)?;
597 if !objects.is_empty() {
598 return Err(SqliteStoreError::UnledgeredDomainObjects {
599 domain: domain.name.to_string(),
600 objects,
601 bridgeable: domain.bridge_eligibility(&tx),
602 });
603 }
604 }
605 let current = current.unwrap_or(0);
606 if current == supported {
607 tx.rollback()?;
608 return Ok(LedgerReport {
609 from_version: current,
610 to_version: current,
611 });
612 }
613
614 if !ledger_table_exists(&tx)? {
617 tx.execute_batch(CREATE_LEDGER_SQL)?;
618 validate_ledger_shape(&tx)?;
619 }
620
621 if current == 0 {
622 tx.execute_batch(CUSTODY_SAVEPOINT_SQL)?;
623 (domain.initialize_current)(&tx).map_err(|source| SqliteStoreError::MigrationFailed {
624 domain: domain.name.to_string(),
625 version: supported,
626 name: "initialize-current".to_string(),
627 source,
628 })?;
629 if tx.is_autocommit() || tx.execute_batch(CUSTODY_RELEASE_SQL).is_err() {
630 return Err(SqliteStoreError::MigrationBrokeTransaction {
631 domain: domain.name.to_string(),
632 version: supported,
633 name: "initialize-current".to_string(),
634 });
635 }
636 } else {
637 for migration in domain.migrations.iter().filter(|m| m.version > current) {
638 tx.execute_batch(CUSTODY_SAVEPOINT_SQL)?;
639 (migration.apply)(&tx).map_err(|source| SqliteStoreError::MigrationFailed {
640 domain: domain.name.to_string(),
641 version: migration.version,
642 name: migration.name.to_string(),
643 source,
644 })?;
645 if tx.is_autocommit() || tx.execute_batch(CUSTODY_RELEASE_SQL).is_err() {
655 return Err(SqliteStoreError::MigrationBrokeTransaction {
656 domain: domain.name.to_string(),
657 version: migration.version,
658 name: migration.name.to_string(),
659 });
660 }
661 }
662 }
663 verify_current_schema_fingerprint(&tx, domain).map_err(|detail| {
664 SqliteStoreError::SchemaFingerprintMismatch {
665 domain: domain.name.to_string(),
666 version: supported,
667 detail,
668 }
669 })?;
670 tx.execute(
671 "INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, ?2)
672 ON CONFLICT(domain) DO UPDATE SET version = excluded.version",
673 rusqlite::params![domain.name, supported],
674 )?;
675 verify_ledger_stamp(&tx, domain.name, supported)?;
676 tx.commit()?;
677
678 Ok(LedgerReport {
679 from_version: current,
680 to_version: supported,
681 })
682}
683
684pub type MaintenancePrepareFn =
686 for<'connection> fn(
687 &Transaction<'connection>,
688 ) -> Result<MaintenancePrepareReport, rusqlite::Error>;
689
690pub fn bridge_unledgered_domain(
715 conn: &mut Connection,
716 domain: &SchemaDomain,
717 target_version: i64,
718 recoverable_source_versions: &[i64],
719 prepare: Option<MaintenancePrepareFn>,
720) -> Result<MaintenanceBridgeReport, SqliteStoreError> {
721 domain.validate()?;
722 let supported = domain.supported_version();
723 if target_version > supported {
724 return Err(SqliteStoreError::SchemaFromTheFuture {
725 domain: domain.name.to_string(),
726 found: target_version,
727 supported,
728 });
729 }
730 if !domain.accepts_existing_version(target_version) {
731 return Err(unsupported_predecessor(domain, target_version));
732 }
733 validate_recoverable_source_versions(domain, target_version, recoverable_source_versions)?;
734
735 let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
736 let current = if ledger_table_exists(&tx)? {
737 validate_ledger_shape(&tx)?;
738 read_version(&tx, domain.name)?
739 } else {
740 None
741 };
742
743 let mut inferred_oracles = None;
744 let from_version = if let Some(found) = current {
745 if found > target_version {
746 return Err(SqliteStoreError::SchemaFromTheFuture {
747 domain: domain.name.to_string(),
748 found,
749 supported: target_version,
750 });
751 }
752 if !domain.accepts_existing_version(found) {
753 return Err(unsupported_predecessor(domain, found));
754 }
755 domain.verify_stamped_version(&tx, found)?;
756 if found == target_version && prepare.is_none() {
757 return Ok(MaintenanceBridgeReport {
758 from_version: found,
759 to_version: found,
760 prepared: 0,
761 refused: Vec::new(),
762 });
763 }
764 found
765 } else {
766 let objects = find_owned_objects(&tx, domain)?;
767 if objects.is_empty() {
768 return Ok(MaintenanceBridgeReport {
769 from_version: 0,
770 to_version: 0,
771 prepared: 0,
772 refused: Vec::new(),
773 });
774 }
775
776 let matched = authenticate_unledgered_source_version(
777 &tx,
778 domain,
779 target_version,
780 recoverable_source_versions,
781 || objects.clone(),
782 )?;
783 inferred_oracles = Some(matched.oracles);
784 matched.version
785 };
786
787 let oracles = match inferred_oracles {
788 Some(oracles) => oracles,
789 None => build_migration_prefix_oracles(domain, target_version)?,
790 };
791 validate_domain_trigger_isolation(&tx, domain, from_version)?;
792
793 let (prepared, refused) = match prepare {
794 Some(prepare) => {
795 let report =
796 run_with_custody(&tx, domain, from_version, "maintenance-prepare", prepare)?;
797 (report.changed, report.refused)
798 }
799 None => (0, Vec::new()),
800 };
801 for migration in domain
802 .migrations
803 .iter()
804 .filter(|migration| migration.version > from_version && migration.version <= target_version)
805 {
806 run_with_custody(
807 &tx,
808 domain,
809 migration.version,
810 migration.name,
811 migration.apply,
812 )?;
813 }
814
815 let target = oracles
816 .iter()
817 .find_map(|(version, fingerprint)| (*version == target_version).then_some(fingerprint))
818 .ok_or_else(|| SqliteStoreError::InvalidMigrationList {
819 domain: domain.name.to_string(),
820 detail: format!(
821 "migration-prefix oracle did not produce requested target version {target_version}"
822 ),
823 })?;
824 let converged = domain_catalog_fingerprint(&tx, domain).map_err(|detail| {
825 SqliteStoreError::SchemaFingerprintMismatch {
826 domain: domain.name.to_string(),
827 version: target_version,
828 detail,
829 }
830 })?;
831 if &converged != target {
832 return Err(SqliteStoreError::SchemaFingerprintMismatch {
833 domain: domain.name.to_string(),
834 version: target_version,
835 detail: format!(
836 "migration-prefix catalog differs: expected {target:?}, found {converged:?}"
837 ),
838 });
839 }
840 domain.verify_predecessor(&tx, target_version)?;
841
842 if current != Some(target_version) {
843 if !ledger_table_exists(&tx)? {
844 tx.execute_batch(CREATE_LEDGER_SQL)?;
845 validate_ledger_shape(&tx)?;
846 }
847 tx.execute(
848 "INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, ?2)
849 ON CONFLICT(domain) DO UPDATE SET version = excluded.version",
850 rusqlite::params![domain.name, target_version],
851 )?;
852 }
853 verify_ledger_stamp(&tx, domain.name, target_version)?;
854 tx.commit()?;
855
856 Ok(MaintenanceBridgeReport {
857 from_version,
858 to_version: target_version,
859 prepared,
860 refused,
861 })
862}
863
864fn validate_recoverable_source_versions(
865 domain: &SchemaDomain,
866 target_version: i64,
867 versions: &[i64],
868) -> Result<(), SqliteStoreError> {
869 let mut previous = None;
870 for &version in versions {
871 if version <= 0 || version > target_version {
872 return Err(SqliteStoreError::InvalidMigrationList {
873 domain: domain.name.to_string(),
874 detail: format!(
875 "recoverable source version {version} is outside 1..={target_version}"
876 ),
877 });
878 }
879 if previous.is_some_and(|prior| prior >= version) {
880 return Err(SqliteStoreError::InvalidMigrationList {
881 domain: domain.name.to_string(),
882 detail: "recoverable source versions must be strictly increasing".to_string(),
883 });
884 }
885 previous = Some(version);
886 }
887 Ok(())
888}
889
890fn verify_ledger_stamp(
891 conn: &Connection,
892 domain: &str,
893 expected: i64,
894) -> Result<(), SqliteStoreError> {
895 let found = read_version(conn, domain)?;
896 if found != Some(expected) {
897 return Err(malformed(format!(
898 "domain `{domain}` stamp did not persist exact version {expected}; found {found:?}"
899 )));
900 }
901 Ok(())
902}
903
904fn validate_domain_trigger_isolation(
905 conn: &Connection,
906 domain: &SchemaDomain,
907 source_version: i64,
908) -> Result<(), SqliteStoreError> {
909 let all_objects = all_domain_objects(domain);
910 let allowed_triggers = all_objects
911 .iter()
912 .filter(|object| object.kind == SchemaObjectKind::Trigger)
913 .map(|object| object.name)
914 .collect::<Vec<_>>();
915 let mut statement = conn
916 .prepare(
917 "SELECT 'main', name FROM main.sqlite_schema
918 WHERE type = 'trigger' AND tbl_name = ?1 COLLATE NOCASE
919 UNION ALL
920 SELECT 'temp', name FROM temp.sqlite_schema
921 WHERE type = 'trigger' AND tbl_name = ?1 COLLATE NOCASE
922 ORDER BY 1, 2",
923 )
924 .map_err(SqliteStoreError::Sqlite)?;
925 let mut refused = Vec::new();
926 for target in all_objects.iter().filter(|object| {
927 matches!(
928 object.kind,
929 SchemaObjectKind::Table | SchemaObjectKind::View
930 )
931 }) {
932 let rows = statement
933 .query_map([target.name], |row| {
934 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
935 })
936 .map_err(SqliteStoreError::Sqlite)?;
937 for row in rows {
938 let (schema, trigger) = row.map_err(SqliteStoreError::Sqlite)?;
939 let declared = schema == "main"
940 && allowed_triggers
941 .iter()
942 .any(|allowed| allowed.eq_ignore_ascii_case(&trigger));
943 if !declared {
944 refused.push(format!("{schema}.{trigger} on {}", target.name));
945 }
946 }
947 }
948 refused.sort();
949 refused.dedup();
950 if !refused.is_empty() {
951 return Err(SqliteStoreError::SchemaFingerprintMismatch {
952 domain: domain.name.to_string(),
953 version: source_version,
954 detail: format!(
955 "undeclared or TEMP triggers can intercept maintenance writes: {refused:?}"
956 ),
957 });
958 }
959 Ok(())
960}
961
962fn run_with_custody<T>(
963 tx: &Transaction<'_>,
964 domain: &SchemaDomain,
965 version: i64,
966 name: &str,
967 body: fn(&Transaction<'_>) -> Result<T, rusqlite::Error>,
968) -> Result<T, SqliteStoreError> {
969 tx.execute_batch(CUSTODY_SAVEPOINT_SQL)?;
970 let body_result = body(tx);
971 if tx.is_autocommit() || tx.execute_batch(CUSTODY_RELEASE_SQL).is_err() {
972 return Err(SqliteStoreError::MigrationBrokeTransaction {
973 domain: domain.name.to_string(),
974 version,
975 name: name.to_string(),
976 });
977 }
978 body_result.map_err(|source| SqliteStoreError::MigrationFailed {
979 domain: domain.name.to_string(),
980 version,
981 name: name.to_string(),
982 source,
983 })
984}
985
986struct AuthenticatedUnledgeredSource {
989 version: i64,
990 oracles: Vec<(i64, DomainCatalogFingerprint)>,
991}
992
993fn authenticate_unledgered_source_version(
1008 conn: &Connection,
1009 domain: &SchemaDomain,
1010 target_version: i64,
1011 recoverable_source_versions: &[i64],
1012 objects: impl FnOnce() -> Vec<String>,
1013) -> Result<AuthenticatedUnledgeredSource, SqliteStoreError> {
1014 if recoverable_source_versions.is_empty() {
1015 return Err(SqliteStoreError::UnledgeredSchemaNoMatch {
1019 domain: domain.name.to_string(),
1020 target_version,
1021 objects: objects(),
1022 });
1023 }
1024 let oracles = build_migration_prefix_oracles(domain, target_version)?;
1025 let actual = domain_catalog_fingerprint(conn, domain).map_err(|detail| {
1026 SqliteStoreError::UnledgeredSchemaNoMatch {
1027 domain: domain.name.to_string(),
1028 target_version,
1029 objects: vec![detail],
1030 }
1031 })?;
1032 let mut matches = oracles
1033 .iter()
1034 .filter_map(|(version, fingerprint)| {
1035 (recoverable_source_versions.contains(version) && fingerprint == &actual)
1036 .then_some(*version)
1037 })
1038 .collect::<Vec<_>>();
1039 for predecessor in domain.released_predecessors.iter().filter(|predecessor| {
1047 recoverable_source_versions.contains(&predecessor.version)
1048 && predecessor.version <= target_version
1049 }) {
1050 if (predecessor.verify)(conn).is_ok() && !matches.contains(&predecessor.version) {
1051 matches.push(predecessor.version);
1052 }
1053 }
1054 matches.sort_unstable();
1055 match matches.as_slice() {
1056 [version] => Ok(AuthenticatedUnledgeredSource {
1057 version: *version,
1058 oracles,
1059 }),
1060 [] => Err(SqliteStoreError::UnledgeredSchemaNoMatch {
1061 domain: domain.name.to_string(),
1062 target_version,
1063 objects: objects(),
1064 }),
1065 _ => Err(SqliteStoreError::UnledgeredSchemaAmbiguous {
1066 domain: domain.name.to_string(),
1067 target_version,
1068 matches,
1069 }),
1070 }
1071}
1072
1073fn build_migration_prefix_oracles(
1074 domain: &SchemaDomain,
1075 target_version: i64,
1076) -> Result<Vec<(i64, DomainCatalogFingerprint)>, SqliteStoreError> {
1077 let mut expected = Connection::open_in_memory().map_err(SqliteStoreError::Sqlite)?;
1078 let tx = expected
1079 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
1080 .map_err(SqliteStoreError::Sqlite)?;
1081 let mut oracles = Vec::with_capacity(target_version as usize);
1082 for migration in domain
1083 .migrations
1084 .iter()
1085 .filter(|migration| migration.version <= target_version)
1086 {
1087 (migration.apply)(&tx).map_err(|source| SqliteStoreError::MigrationFailed {
1088 domain: domain.name.to_string(),
1089 version: migration.version,
1090 name: format!("migration-prefix-oracle:{}", migration.name),
1091 source,
1092 })?;
1093 let fingerprint = domain_catalog_fingerprint(&tx, domain).map_err(|detail| {
1094 SqliteStoreError::InvalidMigrationList {
1095 domain: domain.name.to_string(),
1096 detail: format!(
1097 "migration-prefix oracle version {} is inconsistent with ownership: {detail}",
1098 migration.version
1099 ),
1100 }
1101 })?;
1102 oracles.push((migration.version, fingerprint));
1103 }
1104 Ok(oracles)
1105}
1106
1107#[derive(Debug, PartialEq, Eq)]
1108struct DomainCatalogFingerprint {
1109 names: Vec<(String, String)>,
1110 objects: Vec<CatalogObjectFingerprint>,
1111}
1112
1113fn domain_catalog_fingerprint(
1114 conn: &Connection,
1115 domain: &SchemaDomain,
1116) -> Result<DomainCatalogFingerprint, String> {
1117 let all_objects = all_domain_objects(domain);
1118 let names = catalog_names(conn, &all_objects).map_err(|error| error.to_string())?;
1119 let declared = all_objects
1120 .iter()
1121 .map(|object| (object.name, object))
1122 .collect::<BTreeMap<_, _>>();
1123 let mut objects = Vec::with_capacity(names.len());
1124 for (kind, name) in &names {
1125 let Some(object) = declared.get(name.as_str()) else {
1126 return Err(format!("undeclared owned object `{name}`"));
1127 };
1128 if kind != object.kind.sqlite_name() {
1129 return Err(format!(
1130 "owned object `{name}` has kind `{kind}`, expected `{}`",
1131 object.kind.sqlite_name()
1132 ));
1133 }
1134 objects.push(
1135 catalog_fingerprint(conn, object)
1136 .map_err(|error| format!("fingerprint object `{name}`: {error}"))?,
1137 );
1138 }
1139 Ok(DomainCatalogFingerprint { names, objects })
1140}
1141
1142static EXPECTED_CURRENT_CATALOGS: OnceLock<Mutex<BTreeMap<String, Result<String, String>>>> =
1143 OnceLock::new();
1144
1145fn verify_current_schema_fingerprint(
1149 actual: &Connection,
1150 domain: &SchemaDomain,
1151) -> Result<(), String> {
1152 let expected = {
1153 let cache = EXPECTED_CURRENT_CATALOGS.get_or_init(|| Mutex::new(BTreeMap::new()));
1154 let key = current_catalog_cache_key(domain);
1155 let cached = cache
1156 .lock()
1157 .map_err(|_| "current catalog cache lock is poisoned".to_string())?
1158 .get(&key)
1159 .cloned();
1160 if let Some(cached) = cached {
1161 cached?
1162 } else {
1163 let built = build_current_catalog_fingerprint(domain);
1164 cache
1165 .lock()
1166 .map_err(|_| "current catalog cache lock is poisoned".to_string())?
1167 .insert(key, built.clone());
1168 built?
1169 }
1170 };
1171 let actual = compact_catalog_fingerprint(actual, domain, domain.owned_objects)?;
1172 if actual != expected {
1173 return Err(format!(
1174 "current owned catalog differs: expected {expected}, found {actual}"
1175 ));
1176 }
1177 Ok(())
1178}
1179
1180fn current_catalog_cache_key(domain: &SchemaDomain) -> String {
1185 let mut key = format!(
1186 "{}\u{1f}{}\u{1f}{:x}",
1187 domain.name,
1188 domain.supported_version(),
1189 domain.initialize_current as usize
1190 );
1191 for object in domain.owned_objects {
1192 key.push_str(&format!(
1193 "\u{1e}current:{}:{}",
1194 object.kind.sqlite_name(),
1195 object.name
1196 ));
1197 }
1198 for object in domain.retired_objects {
1199 key.push_str(&format!(
1200 "\u{1e}retired:{}:{}",
1201 object.kind.sqlite_name(),
1202 object.name
1203 ));
1204 }
1205 key
1206}
1207
1208#[derive(Debug, Default)]
1211struct CurrentCatalogDiff {
1212 missing: Vec<String>,
1213 unexpected: Vec<String>,
1214 changed: Vec<String>,
1215}
1216
1217impl CurrentCatalogDiff {
1218 fn is_empty(&self) -> bool {
1219 self.missing.is_empty() && self.unexpected.is_empty() && self.changed.is_empty()
1220 }
1221}
1222
1223fn current_catalog_diff(
1228 actual: &Connection,
1229 domain: &SchemaDomain,
1230) -> Result<CurrentCatalogDiff, String> {
1231 let mut oracle =
1232 Connection::open_in_memory().map_err(|error| format!("open current oracle: {error}"))?;
1233 let tx = oracle
1234 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
1235 .map_err(|error| format!("begin current oracle: {error}"))?;
1236 (domain.initialize_current)(&tx).map_err(|error| format!("build current oracle: {error}"))?;
1237 tx.commit()
1238 .map_err(|error| format!("commit current oracle: {error}"))?;
1239 let expected = domain_catalog_entries(&oracle, domain)?;
1240 let found = domain_catalog_entries(actual, domain)?;
1241 let mut diff = CurrentCatalogDiff::default();
1242 for (object, entry) in &expected {
1243 match found.get(object) {
1244 None => diff.missing.push(object.clone()),
1245 Some(found_entry) if found_entry != entry => diff.changed.push(object.clone()),
1246 Some(_) => {}
1247 }
1248 }
1249 for object in found.keys() {
1250 if !expected.contains_key(object) {
1251 diff.unexpected.push(object.clone());
1252 }
1253 }
1254 Ok(diff)
1255}
1256
1257fn domain_catalog_entries(
1260 conn: &Connection,
1261 domain: &SchemaDomain,
1262) -> Result<BTreeMap<String, String>, String> {
1263 let names = all_domain_objects(domain)
1264 .into_iter()
1265 .map(|object| object.name)
1266 .collect::<std::collections::BTreeSet<_>>();
1267 let mut statement = conn
1268 .prepare(
1269 "SELECT type, name, tbl_name, sql
1270 FROM main.sqlite_schema
1271 WHERE name NOT LIKE 'sqlite_%'
1272 ORDER BY type, name",
1273 )
1274 .map_err(|error| error.to_string())?;
1275 let rows = statement
1276 .query_map([], |row| {
1277 Ok((
1278 row.get::<_, String>(0)?,
1279 row.get::<_, String>(1)?,
1280 row.get::<_, String>(2)?,
1281 row.get::<_, Option<String>>(3)?,
1282 ))
1283 })
1284 .map_err(|error| error.to_string())?;
1285 let mut entries = BTreeMap::new();
1286 for row in rows {
1287 let (kind, name, table_name, sql) = row.map_err(|error| error.to_string())?;
1288 if names.contains(name.as_str()) {
1289 entries.insert(
1290 format!("{kind}:{name}"),
1291 format!(
1292 "{table_name}\u{1f}{}",
1293 sql.map(|sql| normalize_schema_sql(&sql))
1294 .unwrap_or_default()
1295 ),
1296 );
1297 }
1298 }
1299 Ok(entries)
1300}
1301
1302fn build_current_catalog_fingerprint(domain: &SchemaDomain) -> Result<String, String> {
1303 let mut expected =
1304 Connection::open_in_memory().map_err(|error| format!("open current oracle: {error}"))?;
1305 let tx = expected
1306 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
1307 .map_err(|error| format!("begin current oracle: {error}"))?;
1308 (domain.initialize_current)(&tx).map_err(|error| format!("build current oracle: {error}"))?;
1309 tx.commit()
1310 .map_err(|error| format!("commit current oracle: {error}"))?;
1311 compact_catalog_fingerprint(&expected, domain, domain.owned_objects)
1312}
1313
1314fn compact_catalog_fingerprint(
1315 conn: &Connection,
1316 domain: &SchemaDomain,
1317 expected_objects: &[SchemaObject],
1318) -> Result<String, String> {
1319 let all_objects = all_domain_objects(domain);
1320 let owned_by_name = all_objects
1321 .iter()
1322 .map(|object| (object.name, object))
1323 .collect::<BTreeMap<_, _>>();
1324 let current_by_name = expected_objects
1325 .iter()
1326 .map(|object| (object.name, object))
1327 .collect::<BTreeMap<_, _>>();
1328 let mut actual_names = Vec::new();
1329 let mut entries = Vec::with_capacity(expected_objects.len());
1330 let mut statement = conn
1331 .prepare(
1332 "SELECT type, name, tbl_name, sql
1333 FROM main.sqlite_schema
1334 WHERE name NOT LIKE 'sqlite_%'
1335 ORDER BY type, name",
1336 )
1337 .map_err(|error| error.to_string())?;
1338 let rows = statement
1339 .query_map([], |row| {
1340 Ok((
1341 row.get::<_, String>(0)?,
1342 row.get::<_, String>(1)?,
1343 row.get::<_, String>(2)?,
1344 row.get::<_, Option<String>>(3)?,
1345 ))
1346 })
1347 .map_err(|error| error.to_string())?;
1348 for row in rows {
1349 let (kind, name, table_name, sql) = row.map_err(|error| error.to_string())?;
1350 if owned_by_name.contains_key(name.as_str()) {
1351 actual_names.push((kind.clone(), name.clone()));
1352 }
1353 if current_by_name.contains_key(name.as_str()) {
1354 entries.push(format!(
1355 "{kind}\u{1f}{name}\u{1f}{table_name}\u{1f}{}",
1356 sql.map(|sql| normalize_schema_sql(&sql))
1357 .unwrap_or_default()
1358 ));
1359 }
1360 }
1361 actual_names.sort();
1362 let mut expected_names = expected_objects
1363 .iter()
1364 .map(|object| {
1365 (
1366 object.kind.sqlite_name().to_string(),
1367 object.name.to_string(),
1368 )
1369 })
1370 .collect::<Vec<_>>();
1371 expected_names.sort();
1372 if actual_names != expected_names {
1373 return Err(format!(
1374 "owned object set differs: expected {expected_names:?}, found {actual_names:?}"
1375 ));
1376 }
1377
1378 entries.sort();
1379 Ok(entries.join("\u{1e}"))
1380}
1381
1382fn all_domain_objects(domain: &SchemaDomain) -> Vec<SchemaObject> {
1383 domain
1384 .owned_objects
1385 .iter()
1386 .chain(domain.retired_objects)
1387 .copied()
1388 .collect()
1389}
1390
1391pub fn verify_released_schema_fingerprint(
1401 actual: &Connection,
1402 domain: &SchemaDomain,
1403 released_objects: &[SchemaObject],
1404 build_released: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
1405) -> Result<(), String> {
1406 verify_released_schema(
1407 actual,
1408 domain,
1409 released_objects,
1410 build_released,
1411 ReleasedSchemaComparison::ExactText,
1412 )
1413}
1414
1415pub fn verify_released_schema_structure(
1432 actual: &Connection,
1433 domain: &SchemaDomain,
1434 released_objects: &[SchemaObject],
1435 build_released: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
1436) -> Result<(), String> {
1437 verify_released_schema(
1438 actual,
1439 domain,
1440 released_objects,
1441 build_released,
1442 ReleasedSchemaComparison::StructureOnly,
1443 )
1444}
1445
1446#[derive(Clone, Copy)]
1448enum ReleasedSchemaComparison {
1449 ExactText,
1451 StructureOnly,
1453}
1454
1455fn verify_released_schema(
1456 actual: &Connection,
1457 domain: &SchemaDomain,
1458 released_objects: &[SchemaObject],
1459 build_released: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
1460 comparison: ReleasedSchemaComparison,
1461) -> Result<(), String> {
1462 let mut expected = Connection::open_in_memory()
1463 .map_err(|error| format!("open fingerprint oracle: {error}"))?;
1464 let tx = expected
1465 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
1466 .map_err(|error| format!("begin fingerprint oracle: {error}"))?;
1467 build_released(&tx).map_err(|error| format!("build fingerprint oracle: {error}"))?;
1468 tx.commit()
1469 .map_err(|error| format!("commit fingerprint oracle: {error}"))?;
1470
1471 let expected_names = catalog_names(&expected, released_objects)
1472 .map_err(|error| format!("read fingerprint oracle: {error}"))?;
1473 let mut declared_expected = released_objects
1474 .iter()
1475 .map(|object| {
1476 (
1477 object.kind.sqlite_name().to_string(),
1478 object.name.to_string(),
1479 )
1480 })
1481 .collect::<Vec<_>>();
1482 declared_expected.sort();
1483 if expected_names != declared_expected {
1484 return Err(format!(
1485 "frozen builder produced {expected_names:?}, manifest declares {declared_expected:?}"
1486 ));
1487 }
1488
1489 let actual_names = catalog_names(actual, &all_domain_objects(domain))
1490 .map_err(|error| format!("read actual catalog: {error}"))?;
1491 if actual_names != declared_expected {
1492 return Err(format!(
1493 "owned object set differs: expected {declared_expected:?}, found {actual_names:?}"
1494 ));
1495 }
1496
1497 for object in released_objects {
1498 let mut wanted = catalog_fingerprint(&expected, object)
1499 .map_err(|error| format!("fingerprint oracle {}: {error}", object.name))?;
1500 let mut found = catalog_fingerprint(actual, object)
1501 .map_err(|error| format!("fingerprint actual {}: {error}", object.name))?;
1502 if matches!(comparison, ReleasedSchemaComparison::StructureOnly) {
1503 wanted.normalized_sql = None;
1504 found.normalized_sql = None;
1505 }
1506 if found != wanted {
1507 return Err(format!(
1508 "object `{}` differs: expected {wanted:?}, found {found:?}",
1509 object.name
1510 ));
1511 }
1512 }
1513 Ok(())
1514}
1515
1516#[derive(Debug, PartialEq, Eq)]
1517struct CatalogObjectFingerprint {
1518 kind: String,
1519 name: String,
1520 table_name: String,
1521 normalized_sql: Option<String>,
1522 table_columns: Vec<TableColumnFingerprint>,
1523 foreign_keys: Vec<ForeignKeyFingerprint>,
1524 index: Option<IndexFingerprint>,
1525}
1526
1527#[derive(Debug, PartialEq, Eq)]
1528struct TableColumnFingerprint {
1529 cid: i64,
1530 name: String,
1531 declared_type: String,
1532 not_null: bool,
1533 default_value: Option<String>,
1534 primary_key_position: i64,
1535 hidden: i64,
1536}
1537
1538#[derive(Debug, PartialEq, Eq)]
1539struct ForeignKeyFingerprint {
1540 id: i64,
1541 sequence: i64,
1542 target_table: String,
1543 from_column: String,
1544 to_column: Option<String>,
1545 on_update: String,
1546 on_delete: String,
1547 match_clause: String,
1548}
1549
1550#[derive(Debug, PartialEq, Eq)]
1551struct IndexFingerprint {
1552 unique: bool,
1553 origin: String,
1554 partial: bool,
1555 columns: Vec<IndexColumnFingerprint>,
1556}
1557
1558#[derive(Debug, PartialEq, Eq)]
1559struct IndexColumnFingerprint {
1560 sequence: i64,
1561 column_id: i64,
1562 name: Option<String>,
1563 descending: bool,
1564 collation: Option<String>,
1565 key: bool,
1566}
1567
1568fn catalog_names(
1569 conn: &Connection,
1570 objects: &[SchemaObject],
1571) -> Result<Vec<(String, String)>, rusqlite::Error> {
1572 let mut found = Vec::new();
1573 let mut statement = conn.prepare(
1574 "SELECT type, name FROM main.sqlite_schema
1575 WHERE name = ?1 AND name NOT LIKE 'sqlite_%'
1576 ORDER BY type, name",
1577 )?;
1578 for object in objects {
1579 let rows = statement.query_map([object.name], |row| Ok((row.get(0)?, row.get(1)?)))?;
1580 found.extend(rows.collect::<Result<Vec<_>, _>>()?);
1581 }
1582 found.sort();
1583 found.dedup();
1584 Ok(found)
1585}
1586
1587fn catalog_fingerprint(
1588 conn: &Connection,
1589 object: &SchemaObject,
1590) -> Result<CatalogObjectFingerprint, rusqlite::Error> {
1591 let (kind, name, table_name, sql): (String, String, String, Option<String>) = conn.query_row(
1592 "SELECT type, name, tbl_name, sql FROM main.sqlite_schema WHERE name = ?1",
1593 [object.name],
1594 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
1595 )?;
1596 let (table_columns, foreign_keys) = if kind == "table" || kind == "view" {
1597 (
1598 table_columns(conn, object.name)?,
1599 foreign_keys(conn, object.name)?,
1600 )
1601 } else {
1602 (Vec::new(), Vec::new())
1603 };
1604 let index = if kind == "index" {
1605 Some(index_fingerprint(conn, &table_name, object.name)?)
1606 } else {
1607 None
1608 };
1609 Ok(CatalogObjectFingerprint {
1610 kind,
1611 name,
1612 table_name,
1613 normalized_sql: sql.map(|sql| normalize_schema_sql(&sql)),
1614 table_columns,
1615 foreign_keys,
1616 index,
1617 })
1618}
1619
1620fn table_columns(
1621 conn: &Connection,
1622 table: &str,
1623) -> Result<Vec<TableColumnFingerprint>, rusqlite::Error> {
1624 let mut statement = conn.prepare(
1625 "SELECT cid, name, type, \"notnull\", dflt_value, pk, hidden
1626 FROM pragma_table_xinfo(?1, 'main')
1627 ORDER BY cid",
1628 )?;
1629 let rows = statement.query_map([table], |row| {
1630 Ok(TableColumnFingerprint {
1631 cid: row.get(0)?,
1632 name: row.get(1)?,
1633 declared_type: row.get(2)?,
1634 not_null: row.get(3)?,
1635 default_value: row.get(4)?,
1636 primary_key_position: row.get(5)?,
1637 hidden: row.get(6)?,
1638 })
1639 })?;
1640 rows.collect()
1641}
1642
1643fn foreign_keys(
1644 conn: &Connection,
1645 table: &str,
1646) -> Result<Vec<ForeignKeyFingerprint>, rusqlite::Error> {
1647 let mut statement = conn.prepare(
1648 "SELECT id, seq, \"table\", \"from\", \"to\", on_update, on_delete, \"match\"
1649 FROM pragma_foreign_key_list(?1, 'main')
1650 ORDER BY id, seq",
1651 )?;
1652 let rows = statement.query_map([table], |row| {
1653 Ok(ForeignKeyFingerprint {
1654 id: row.get(0)?,
1655 sequence: row.get(1)?,
1656 target_table: row.get(2)?,
1657 from_column: row.get(3)?,
1658 to_column: row.get(4)?,
1659 on_update: row.get(5)?,
1660 on_delete: row.get(6)?,
1661 match_clause: row.get(7)?,
1662 })
1663 })?;
1664 rows.collect()
1665}
1666
1667fn index_fingerprint(
1668 conn: &Connection,
1669 table: &str,
1670 index: &str,
1671) -> Result<IndexFingerprint, rusqlite::Error> {
1672 let (unique, origin, partial): (bool, String, bool) = conn.query_row(
1673 "SELECT \"unique\", origin, partial
1674 FROM pragma_index_list(?1, 'main')
1675 WHERE name = ?2",
1676 rusqlite::params![table, index],
1677 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
1678 )?;
1679 let mut statement = conn.prepare(
1680 "SELECT seqno, cid, name, desc, coll, key
1681 FROM pragma_index_xinfo(?1, 'main')
1682 ORDER BY seqno",
1683 )?;
1684 let columns = statement
1685 .query_map([index], |row| {
1686 Ok(IndexColumnFingerprint {
1687 sequence: row.get(0)?,
1688 column_id: row.get(1)?,
1689 name: row.get(2)?,
1690 descending: row.get(3)?,
1691 collation: row.get(4)?,
1692 key: row.get(5)?,
1693 })
1694 })?
1695 .collect::<Result<Vec<_>, _>>()?;
1696 Ok(IndexFingerprint {
1697 unique,
1698 origin,
1699 partial,
1700 columns,
1701 })
1702}
1703
1704fn normalize_schema_sql(sql: &str) -> String {
1705 #[derive(Clone, Copy)]
1706 enum LexState {
1707 Normal,
1708 SingleQuoted,
1709 DoubleQuoted,
1710 BacktickQuoted,
1711 BracketQuoted,
1712 LineComment,
1713 BlockComment,
1714 }
1715
1716 let bytes = sql.as_bytes();
1717 let mut collapsed = Vec::with_capacity(bytes.len());
1718 let mut state = LexState::Normal;
1719 let mut index = 0;
1720 while index < bytes.len() {
1721 let byte = bytes[index];
1722 match state {
1723 LexState::Normal => {
1724 if byte.is_ascii_whitespace() {
1725 if collapsed.last().is_some_and(|last| *last != b' ') {
1726 collapsed.push(b' ');
1727 }
1728 } else {
1729 collapsed.push(byte);
1730 state = match byte {
1731 b'\'' => LexState::SingleQuoted,
1732 b'"' => LexState::DoubleQuoted,
1733 b'`' => LexState::BacktickQuoted,
1734 b'[' => LexState::BracketQuoted,
1735 b'-' if bytes.get(index + 1) == Some(&b'-') => LexState::LineComment,
1736 b'/' if bytes.get(index + 1) == Some(&b'*') => LexState::BlockComment,
1737 _ => LexState::Normal,
1738 };
1739 }
1740 }
1741 LexState::SingleQuoted | LexState::DoubleQuoted | LexState::BacktickQuoted => {
1742 collapsed.push(byte);
1743 let delimiter = match state {
1744 LexState::SingleQuoted => b'\'',
1745 LexState::DoubleQuoted => b'"',
1746 LexState::BacktickQuoted => b'`',
1747 _ => unreachable!(),
1748 };
1749 if byte == delimiter {
1750 if bytes.get(index + 1) == Some(&delimiter) {
1751 index += 1;
1752 collapsed.push(delimiter);
1753 } else {
1754 state = LexState::Normal;
1755 }
1756 }
1757 }
1758 LexState::BracketQuoted => {
1759 collapsed.push(byte);
1760 if byte == b']' {
1761 if bytes.get(index + 1) == Some(&b']') {
1762 index += 1;
1763 collapsed.push(b']');
1764 } else {
1765 state = LexState::Normal;
1766 }
1767 }
1768 }
1769 LexState::LineComment => {
1770 collapsed.push(byte);
1771 if byte == b'\n' || byte == b'\r' {
1772 state = LexState::Normal;
1773 }
1774 }
1775 LexState::BlockComment => {
1776 collapsed.push(byte);
1777 if byte == b'*' && bytes.get(index + 1) == Some(&b'/') {
1778 index += 1;
1779 collapsed.push(b'/');
1780 state = LexState::Normal;
1781 }
1782 }
1783 }
1784 index += 1;
1785 }
1786
1787 const IF_NOT_EXISTS: &[u8] = b"IF NOT EXISTS";
1788 let mut normalized = Vec::with_capacity(collapsed.len());
1789 let mut state = LexState::Normal;
1790 let mut index = 0;
1791 while index < collapsed.len() {
1792 let byte = collapsed[index];
1793 if matches!(state, LexState::Normal)
1794 && collapsed
1795 .get(index..index + IF_NOT_EXISTS.len())
1796 .is_some_and(|candidate| candidate.eq_ignore_ascii_case(IF_NOT_EXISTS))
1797 && (index == 0 || !is_sql_identifier_byte(collapsed[index - 1]))
1798 && collapsed
1799 .get(index + IF_NOT_EXISTS.len())
1800 .is_none_or(|after| !is_sql_identifier_byte(*after))
1801 {
1802 index += IF_NOT_EXISTS.len();
1803 if collapsed.get(index) == Some(&b' ') {
1804 index += 1;
1805 }
1806 continue;
1807 }
1808 normalized.push(byte);
1809 match state {
1810 LexState::Normal => {
1811 state = match byte {
1812 b'\'' => LexState::SingleQuoted,
1813 b'"' => LexState::DoubleQuoted,
1814 b'`' => LexState::BacktickQuoted,
1815 b'[' => LexState::BracketQuoted,
1816 b'-' if collapsed.get(index + 1) == Some(&b'-') => LexState::LineComment,
1817 b'/' if collapsed.get(index + 1) == Some(&b'*') => LexState::BlockComment,
1818 _ => LexState::Normal,
1819 };
1820 }
1821 LexState::SingleQuoted | LexState::DoubleQuoted | LexState::BacktickQuoted => {
1822 let delimiter = match state {
1823 LexState::SingleQuoted => b'\'',
1824 LexState::DoubleQuoted => b'"',
1825 LexState::BacktickQuoted => b'`',
1826 _ => unreachable!(),
1827 };
1828 if byte == delimiter {
1829 if collapsed.get(index + 1) == Some(&delimiter) {
1830 index += 1;
1831 normalized.push(delimiter);
1832 } else {
1833 state = LexState::Normal;
1834 }
1835 }
1836 }
1837 LexState::BracketQuoted => {
1838 if byte == b']' {
1839 if collapsed.get(index + 1) == Some(&b']') {
1840 index += 1;
1841 normalized.push(b']');
1842 } else {
1843 state = LexState::Normal;
1844 }
1845 }
1846 }
1847 LexState::LineComment => {
1848 if byte == b'\n' || byte == b'\r' {
1849 state = LexState::Normal;
1850 }
1851 }
1852 LexState::BlockComment => {
1853 if byte == b'*' && collapsed.get(index + 1) == Some(&b'/') {
1854 index += 1;
1855 normalized.push(b'/');
1856 state = LexState::Normal;
1857 }
1858 }
1859 }
1860 index += 1;
1861 }
1862 String::from_utf8(normalized).unwrap_or_else(|_| sql.to_string())
1863}
1864
1865fn is_sql_identifier_byte(byte: u8) -> bool {
1866 byte.is_ascii_alphanumeric() || byte == b'_'
1867}
1868
1869fn unsupported_predecessor(domain: &SchemaDomain, found: i64) -> SqliteStoreError {
1870 SqliteStoreError::UnsupportedSchemaPredecessor {
1871 domain: domain.name.to_string(),
1872 found,
1873 supported: domain.supported_version(),
1874 allowed: domain.allowed_existing_versions.to_vec(),
1875 }
1876}
1877
1878fn find_owned_objects(
1883 conn: &Connection,
1884 domain: &SchemaDomain,
1885) -> Result<Vec<String>, SqliteStoreError> {
1886 let mut found = Vec::new();
1887 let mut stmt = conn.prepare(
1888 "SELECT type, name FROM main.sqlite_schema
1889 WHERE name = ?1 AND name NOT LIKE 'sqlite_%'
1890 ORDER BY type, name",
1891 )?;
1892 for expected in domain.owned_objects.iter().chain(domain.retired_objects) {
1893 let rows = stmt.query_map([expected.name], |row| {
1894 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
1895 })?;
1896 for row in rows {
1897 let (actual_kind, name) = row?;
1898 found.push(format!(
1899 "{actual_kind}:{name} (expected {})",
1900 expected.kind.sqlite_name()
1901 ));
1902 }
1903 }
1904 found.sort();
1905 found.dedup();
1906 Ok(found)
1907}
1908
1909fn ledger_table_exists(conn: &Connection) -> Result<bool, SqliteStoreError> {
1910 let exists = conn
1911 .query_row(
1912 "SELECT 1 FROM main.sqlite_master WHERE type = 'table' AND name = 'meerkat_schema'",
1913 [],
1914 |_| Ok(()),
1915 )
1916 .optional()?
1917 .is_some();
1918 Ok(exists)
1919}
1920
1921fn malformed(detail: String) -> SqliteStoreError {
1922 SqliteStoreError::LedgerMalformed { detail }
1923}
1924
1925fn validate_ledger_shape(conn: &Connection) -> Result<(), SqliteStoreError> {
1934 let mut stmt = conn.prepare(
1935 "SELECT name, type, \"notnull\", pk FROM pragma_table_info('meerkat_schema', 'main')",
1936 )?;
1937 let mut rows = stmt.query([])?;
1938 let mut domain_ok = false;
1939 let mut version_ok = false;
1940 while let Some(row) = rows.next()? {
1941 let name: String = row.get(0)?;
1942 let decl_type: String = row.get(1)?;
1943 let notnull: bool = row.get(2)?;
1944 let pk: i64 = row.get(3)?;
1945 match name.as_str() {
1946 "domain" => {
1947 if !decl_type.eq_ignore_ascii_case("TEXT") || pk != 1 {
1948 return Err(malformed(format!(
1949 "column `domain` must be `TEXT PRIMARY KEY`, found type `{decl_type}` \
1950 with pk position {pk}"
1951 )));
1952 }
1953 domain_ok = true;
1954 }
1955 "version" => {
1956 if !decl_type.eq_ignore_ascii_case("INTEGER") || !notnull || pk != 0 {
1957 return Err(malformed(format!(
1958 "column `version` must be non-key `INTEGER NOT NULL`, found type \
1959 `{decl_type}` notnull={notnull} pk position {pk}"
1960 )));
1961 }
1962 version_ok = true;
1963 }
1964 other => {
1965 if pk != 0 {
1966 return Err(malformed(format!(
1967 "unexpected primary-key column `{other}`"
1968 )));
1969 }
1970 }
1971 }
1972 }
1973 if !domain_ok || !version_ok {
1974 return Err(malformed(
1975 "table lacks the pinned `domain`/`version` columns".to_string(),
1976 ));
1977 }
1978 let mut trigger_stmt = conn.prepare(
1979 "SELECT 'main', name FROM main.sqlite_schema
1980 WHERE type = 'trigger' AND tbl_name = 'meerkat_schema' COLLATE NOCASE
1981 UNION ALL
1982 SELECT 'temp', name FROM temp.sqlite_schema
1983 WHERE type = 'trigger' AND tbl_name = 'meerkat_schema' COLLATE NOCASE
1984 ORDER BY 1, 2",
1985 )?;
1986 let triggers = trigger_stmt
1987 .query_map([], |row| {
1988 Ok(format!(
1989 "{}.{}",
1990 row.get::<_, String>(0)?,
1991 row.get::<_, String>(1)?
1992 ))
1993 })?
1994 .collect::<Result<Vec<_>, _>>()?;
1995 if !triggers.is_empty() {
1996 return Err(malformed(format!(
1997 "table has attached triggers {triggers:?}; ledger writes must be isolated"
1998 )));
1999 }
2000 Ok(())
2001}
2002
2003fn read_version(conn: &Connection, domain: &str) -> Result<Option<i64>, SqliteStoreError> {
2010 let mut stmt = conn.prepare("SELECT version FROM main.meerkat_schema WHERE domain = ?1")?;
2011 let mut rows = stmt.query([domain])?;
2012 let Some(row) = rows.next()? else {
2013 return Ok(None);
2014 };
2015 let version: i64 = row.get(0)?;
2016 if rows.next()?.is_some() {
2017 return Err(malformed(format!(
2018 "multiple ledger rows for domain `{domain}`"
2019 )));
2020 }
2021 if version <= 0 {
2022 return Err(malformed(format!(
2023 "domain `{domain}` records non-positive version {version}"
2024 )));
2025 }
2026 Ok(Some(version))
2027}
2028
2029#[cfg(test)]
2030#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
2031mod tests {
2032 use super::*;
2033 use crate::profile::{ConnectionProfile, open};
2034
2035 fn create_t1(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2036 tx.execute_batch("CREATE TABLE IF NOT EXISTS t1 (x INTEGER)")
2037 }
2038
2039 fn add_column_guarded(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2040 let has_column = tx
2041 .prepare("PRAGMA table_info(t1)")?
2042 .query_map([], |row| row.get::<_, String>(1))?
2043 .collect::<Result<Vec<_>, _>>()?
2044 .iter()
2045 .any(|name| name == "y");
2046 if !has_column {
2047 tx.execute_batch("ALTER TABLE t1 ADD COLUMN y TEXT")?;
2048 }
2049 Ok(())
2050 }
2051
2052 fn initialize_v2(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2053 create_t1(tx)?;
2054 add_column_guarded(tx)
2055 }
2056
2057 fn initialize_v2_alt(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2058 tx.execute_batch("CREATE TABLE t1 (x INTEGER, z BLOB)")
2059 }
2060
2061 const RELEASED_V1_OBJECTS: &[SchemaObject] = &[SchemaObject {
2062 kind: SchemaObjectKind::Table,
2063 name: "t1",
2064 }];
2065
2066 fn verify_v1(conn: &Connection) -> Result<(), String> {
2067 verify_released_schema_fingerprint(conn, &DOMAIN_V2, RELEASED_V1_OBJECTS, create_t1)
2068 }
2069
2070 const DOMAIN_V1: SchemaDomain = SchemaDomain {
2071 name: "test-domain",
2072 migrations: &[Migration {
2073 version: 1,
2074 name: "base",
2075 apply: create_t1,
2076 }],
2077 initialize_current: create_t1,
2078 allowed_existing_versions: &[1],
2079 bridge_recoverable_versions: &[1],
2080 released_predecessors: &[],
2081 owned_objects: &[SchemaObject {
2082 kind: SchemaObjectKind::Table,
2083 name: "t1",
2084 }],
2085 retired_objects: &[],
2086 };
2087
2088 const DOMAIN_V2: SchemaDomain = SchemaDomain {
2089 name: "test-domain",
2090 migrations: &[
2091 Migration {
2092 version: 1,
2093 name: "base",
2094 apply: create_t1,
2095 },
2096 Migration {
2097 version: 2,
2098 name: "add-y",
2099 apply: add_column_guarded,
2100 },
2101 ],
2102 initialize_current: initialize_v2,
2103 allowed_existing_versions: &[1, 2],
2104 bridge_recoverable_versions: &[1],
2105 released_predecessors: &[SchemaPredecessor {
2106 version: 1,
2107 verify: verify_v1,
2108 }],
2109 owned_objects: &[SchemaObject {
2110 kind: SchemaObjectKind::Table,
2111 name: "t1",
2112 }],
2113 retired_objects: &[],
2114 };
2115
2116 const DOMAIN_V2_ALT_INITIALIZER: SchemaDomain = SchemaDomain {
2117 name: "test-domain",
2118 migrations: &[
2119 Migration {
2120 version: 1,
2121 name: "base",
2122 apply: create_t1,
2123 },
2124 Migration {
2125 version: 2,
2126 name: "alt-current",
2127 apply: add_column_guarded,
2128 },
2129 ],
2130 initialize_current: initialize_v2_alt,
2131 allowed_existing_versions: &[2],
2132 bridge_recoverable_versions: &[1],
2133 released_predecessors: &[],
2134 owned_objects: &[SchemaObject {
2135 kind: SchemaObjectKind::Table,
2136 name: "t1",
2137 }],
2138 retired_objects: &[],
2139 };
2140
2141 fn temp_conn(dir: &tempfile::TempDir) -> Connection {
2142 open(&dir.path().join("db.sqlite3"), ConnectionProfile::PRIMARY).expect("open")
2143 }
2144
2145 #[test]
2146 fn structure_verifier_ignores_create_text_but_rejects_shape_drift() {
2147 const STRUCTURE_OBJECTS: &[SchemaObject] = &[
2148 SchemaObject {
2149 kind: SchemaObjectKind::Table,
2150 name: "t1",
2151 },
2152 SchemaObject {
2153 kind: SchemaObjectKind::Index,
2154 name: "t1_idx",
2155 },
2156 ];
2157 fn build_frozen(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2158 tx.execute_batch(
2159 "CREATE TABLE IF NOT EXISTS t1 (x INTEGER NOT NULL);
2160 CREATE INDEX IF NOT EXISTS t1_idx ON t1(x DESC);",
2161 )
2162 }
2163 const STRUCTURE_DOMAIN: SchemaDomain = SchemaDomain {
2164 name: "structure-domain",
2165 migrations: &[Migration {
2166 version: 1,
2167 name: "base",
2168 apply: build_frozen,
2169 }],
2170 initialize_current: build_frozen,
2171 allowed_existing_versions: &[1],
2172 bridge_recoverable_versions: &[1],
2173 released_predecessors: &[],
2174 owned_objects: STRUCTURE_OBJECTS,
2175 retired_objects: &[],
2176 };
2177 let make = |ddl: &str| {
2178 let conn = Connection::open_in_memory().expect("in-memory fixture");
2179 conn.execute_batch(ddl).expect("fixture ddl");
2180 conn
2181 };
2182
2183 let quoted = make(
2187 "CREATE TABLE \"t1\" (x INTEGER NOT NULL);
2188 CREATE INDEX t1_idx ON t1(x DESC);",
2189 );
2190 verify_released_schema_structure(
2191 "ed,
2192 &STRUCTURE_DOMAIN,
2193 STRUCTURE_OBJECTS,
2194 build_frozen,
2195 )
2196 .expect("CREATE-text drift alone must pass the structural verifier");
2197 verify_released_schema_fingerprint(
2198 "ed,
2199 &STRUCTURE_DOMAIN,
2200 STRUCTURE_OBJECTS,
2201 build_frozen,
2202 )
2203 .expect_err("the exact-text verifier must still reject the same catalog");
2204
2205 let wrong_type = make(
2206 "CREATE TABLE t1 (x TEXT NOT NULL);
2207 CREATE INDEX t1_idx ON t1(x DESC);",
2208 );
2209 verify_released_schema_structure(
2210 &wrong_type,
2211 &STRUCTURE_DOMAIN,
2212 STRUCTURE_OBJECTS,
2213 build_frozen,
2214 )
2215 .expect_err("column type drift must fail structurally");
2216
2217 let wrong_index = make(
2218 "CREATE TABLE t1 (x INTEGER NOT NULL);
2219 CREATE INDEX t1_idx ON t1(x ASC);",
2220 );
2221 verify_released_schema_structure(
2222 &wrong_index,
2223 &STRUCTURE_DOMAIN,
2224 STRUCTURE_OBJECTS,
2225 build_frozen,
2226 )
2227 .expect_err("index direction drift must fail structurally");
2228
2229 let missing_index = make("CREATE TABLE t1 (x INTEGER NOT NULL);");
2230 verify_released_schema_structure(
2231 &missing_index,
2232 &STRUCTURE_DOMAIN,
2233 STRUCTURE_OBJECTS,
2234 build_frozen,
2235 )
2236 .expect_err("a missing owned index must fail the object-set check");
2237 }
2238
2239 #[test]
2240 fn fresh_file_initializes_current_and_stamps() {
2241 let dir = tempfile::tempdir().expect("tempdir");
2242 let mut conn = temp_conn(&dir);
2243 let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("apply");
2244 assert_eq!(
2245 report,
2246 LedgerReport {
2247 from_version: 0,
2248 to_version: 2
2249 }
2250 );
2251 assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(2));
2252 conn.execute("INSERT INTO t1 (x, y) VALUES (1, 'a')", [])
2253 .expect("schema converged");
2254 }
2255
2256 #[test]
2257 fn second_open_is_current_noop() {
2258 let dir = tempfile::tempdir().expect("tempdir");
2259 let mut conn = temp_conn(&dir);
2260 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("first");
2261 let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("second");
2262 assert!(!report.migrated());
2263 }
2264
2265 #[test]
2266 fn current_oracle_cache_binds_initializer_and_manifest_not_only_name_version() {
2267 let first_dir = tempfile::tempdir().expect("first tempdir");
2268 let mut first = temp_conn(&first_dir);
2269 apply_domain_migrations(&mut first, &DOMAIN_V2).expect("first current");
2270
2271 let second_dir = tempfile::tempdir().expect("second tempdir");
2272 let mut second = temp_conn(&second_dir);
2273 apply_domain_migrations(&mut second, &DOMAIN_V2_ALT_INITIALIZER).expect("alt current");
2274 let columns: Vec<String> = second
2275 .prepare("PRAGMA table_info(t1)")
2276 .expect("prepare")
2277 .query_map([], |row| row.get(1))
2278 .expect("query")
2279 .collect::<Result<_, _>>()
2280 .expect("columns");
2281 assert_eq!(columns, vec!["x", "z"]);
2282 }
2283
2284 #[test]
2285 fn current_row_with_partial_catalog_is_refused_before_noop() {
2286 let dir = tempfile::tempdir().expect("tempdir");
2287 let mut conn = temp_conn(&dir);
2288 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("current");
2289 conn.execute_batch("ALTER TABLE t1 ADD COLUMN candidate_partial TEXT")
2290 .expect("partial candidate mutation");
2291 let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse current shape");
2292 match err {
2293 SqliteStoreError::CurrentSchemaMismatch {
2294 version: 2,
2295 missing_objects,
2296 unexpected_objects,
2297 changed_objects,
2298 ..
2299 } => {
2300 assert!(missing_objects.is_empty(), "{missing_objects:?}");
2301 assert!(unexpected_objects.is_empty(), "{unexpected_objects:?}");
2302 assert_eq!(changed_objects, vec!["table:t1".to_string()]);
2303 }
2304 other => panic!("expected a typed current-schema mismatch, got {other:?}"),
2305 }
2306 assert_eq!(
2307 domain_version(&conn, DOMAIN_V2.name).expect("ledger"),
2308 Some(2)
2309 );
2310 }
2311
2312 #[test]
2313 fn current_row_without_its_owned_objects_names_them() {
2314 let dir = tempfile::tempdir().expect("tempdir");
2317 let mut conn = temp_conn(&dir);
2318 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("current");
2319 conn.execute_batch("DROP TABLE t1; CREATE TABLE unrelated_candidate (x INTEGER)")
2320 .expect("foreign-shaped current");
2321 let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse current row");
2322 match err {
2323 SqliteStoreError::CurrentSchemaMismatch {
2324 version: 2,
2325 missing_objects,
2326 unexpected_objects,
2327 changed_objects,
2328 ..
2329 } => {
2330 assert_eq!(missing_objects, vec!["table:t1".to_string()]);
2331 assert!(unexpected_objects.is_empty(), "{unexpected_objects:?}");
2333 assert!(changed_objects.is_empty(), "{changed_objects:?}");
2334 }
2335 other => panic!("expected a typed current-schema mismatch, got {other:?}"),
2336 }
2337 }
2338
2339 #[test]
2340 fn upgrade_applies_only_pending() {
2341 let dir = tempfile::tempdir().expect("tempdir");
2342 let mut conn = temp_conn(&dir);
2343 apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("v1");
2344 let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("v2");
2345 assert_eq!(
2346 report,
2347 LedgerReport {
2348 from_version: 1,
2349 to_version: 2
2350 }
2351 );
2352 }
2353
2354 #[test]
2355 fn allowed_version_with_wrong_catalog_is_refused_without_migration() {
2356 let dir = tempfile::tempdir().expect("tempdir");
2357 let mut conn = temp_conn(&dir);
2358 apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("v1");
2359 conn.execute_batch("ALTER TABLE t1 ADD COLUMN candidate_only TEXT")
2360 .expect("candidate shape");
2361 let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse fingerprint");
2362 assert!(matches!(
2363 err,
2364 SqliteStoreError::SchemaFingerprintMismatch { version: 1, .. }
2365 ));
2366 assert_eq!(
2367 domain_version(&conn, DOMAIN_V1.name).expect("ledger"),
2368 Some(1),
2369 "fingerprint refusal advanced the ledger"
2370 );
2371 let columns: Vec<String> = conn
2372 .prepare("PRAGMA table_info(t1)")
2373 .expect("prepare")
2374 .query_map([], |row| row.get(1))
2375 .expect("query")
2376 .collect::<Result<_, _>>()
2377 .expect("columns");
2378 assert_eq!(columns, vec!["x", "candidate_only"]);
2379 }
2380
2381 #[test]
2382 fn pre_floor_and_gap_versions_are_refused_without_mutation() {
2383 fn no_op(_tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2384 Ok(())
2385 }
2386 fn verify_v2(_conn: &Connection) -> Result<(), String> {
2387 Ok(())
2388 }
2389 const DOMAIN_V3_FLOOR_2: SchemaDomain = SchemaDomain {
2390 name: "floor-domain",
2391 migrations: &[
2392 Migration {
2393 version: 1,
2394 name: "base",
2395 apply: no_op,
2396 },
2397 Migration {
2398 version: 2,
2399 name: "released-floor",
2400 apply: no_op,
2401 },
2402 Migration {
2403 version: 3,
2404 name: "current",
2405 apply: no_op,
2406 },
2407 ],
2408 initialize_current: no_op,
2409 allowed_existing_versions: &[2, 3],
2410 bridge_recoverable_versions: &[1],
2411 released_predecessors: &[SchemaPredecessor {
2412 version: 2,
2413 verify: verify_v2,
2414 }],
2415 owned_objects: &[],
2416 retired_objects: &[],
2417 };
2418 const DOMAIN_V4_GAP_3: SchemaDomain = SchemaDomain {
2419 name: "gap-domain",
2420 migrations: &[
2421 Migration {
2422 version: 1,
2423 name: "old",
2424 apply: no_op,
2425 },
2426 Migration {
2427 version: 2,
2428 name: "released-floor",
2429 apply: no_op,
2430 },
2431 Migration {
2432 version: 3,
2433 name: "unreleased-candidate",
2434 apply: no_op,
2435 },
2436 Migration {
2437 version: 4,
2438 name: "current",
2439 apply: no_op,
2440 },
2441 ],
2442 initialize_current: no_op,
2443 allowed_existing_versions: &[2, 4],
2444 bridge_recoverable_versions: &[1],
2445 released_predecessors: &[SchemaPredecessor {
2446 version: 2,
2447 verify: verify_v2,
2448 }],
2449 owned_objects: &[],
2450 retired_objects: &[],
2451 };
2452 let dir = tempfile::tempdir().expect("tempdir");
2453 let mut conn = temp_conn(&dir);
2454 conn.execute_batch(CREATE_LEDGER_SQL).expect("ledger");
2455 conn.execute(
2456 "INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, 1)",
2457 [DOMAIN_V3_FLOOR_2.name],
2458 )
2459 .expect("pre-floor row");
2460 let err =
2461 apply_domain_migrations(&mut conn, &DOMAIN_V3_FLOOR_2).expect_err("refuse pre-floor");
2462 assert!(matches!(
2463 err,
2464 SqliteStoreError::UnsupportedSchemaPredecessor { found: 1, .. }
2465 ));
2466 assert_eq!(
2467 domain_version(&conn, DOMAIN_V3_FLOOR_2.name).expect("ledger"),
2468 Some(1)
2469 );
2470
2471 conn.execute(
2472 "INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, 3)",
2473 [DOMAIN_V4_GAP_3.name],
2474 )
2475 .expect("gap row");
2476 let err = apply_domain_migrations(&mut conn, &DOMAIN_V4_GAP_3).expect_err("refuse gap");
2477 assert!(matches!(
2478 err,
2479 SqliteStoreError::UnsupportedSchemaPredecessor { found: 3, .. }
2480 ));
2481 assert_eq!(
2482 domain_version(&conn, DOMAIN_V4_GAP_3.name).expect("ledger"),
2483 Some(3)
2484 );
2485 }
2486
2487 #[test]
2488 fn unledgered_owned_objects_are_refused_without_mutation() {
2489 let dir = tempfile::tempdir().expect("tempdir");
2490 let mut conn = temp_conn(&dir);
2491 conn.execute_batch("CREATE TABLE t1 (x INTEGER)")
2492 .expect("unknown unledgered ddl");
2493 let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse");
2494 assert!(matches!(
2495 err,
2496 SqliteStoreError::UnledgeredDomainObjects { .. }
2497 ));
2498 assert!(
2499 !ledger_table_exists(&conn).expect("ledger presence"),
2500 "eligibility refusal must not create the ledger"
2501 );
2502 let columns: Vec<String> = conn
2503 .prepare("PRAGMA table_info(t1)")
2504 .expect("prepare")
2505 .query_map([], |row| row.get(1))
2506 .expect("query")
2507 .collect::<Result<_, _>>()
2508 .expect("columns");
2509 assert_eq!(columns, vec!["x"], "refusal mutated unknown schema");
2510 }
2511
2512 #[test]
2513 fn maintenance_bridge_authenticates_exact_v1_and_migrates_to_v2() {
2514 let dir = tempfile::tempdir().expect("tempdir");
2515 let mut conn = temp_conn(&dir);
2516 conn.execute_batch("CREATE TABLE t1 (x INTEGER)")
2517 .expect("historical unledgered v1");
2518
2519 let report =
2520 bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], None).expect("bridge");
2521 assert_eq!(
2522 report,
2523 MaintenanceBridgeReport {
2524 from_version: 1,
2525 to_version: 2,
2526 prepared: 0,
2527 refused: Vec::new(),
2528 }
2529 );
2530 assert_eq!(
2531 domain_version(&conn, DOMAIN_V2.name).expect("ledger"),
2532 Some(2)
2533 );
2534 let columns = conn
2535 .prepare("PRAGMA table_info(t1)")
2536 .expect("prepare")
2537 .query_map([], |row| row.get::<_, String>(1))
2538 .expect("query")
2539 .collect::<Result<Vec<_>, _>>()
2540 .expect("columns");
2541 assert_eq!(columns, vec!["x", "y"]);
2542
2543 let second = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], None)
2544 .expect("idempotent bridge");
2545 assert_eq!(
2546 second,
2547 MaintenanceBridgeReport {
2548 from_version: 2,
2549 to_version: 2,
2550 prepared: 0,
2551 refused: Vec::new(),
2552 }
2553 );
2554 }
2555
2556 #[test]
2557 fn maintenance_bridge_prepares_and_upgrades_existing_v1_row() {
2558 fn normalize_existing(
2559 tx: &Transaction<'_>,
2560 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2561 let changed = tx.execute("UPDATE t1 SET x = x + 1", [])?;
2562 Ok(MaintenancePrepareReport::rewrote(changed))
2563 }
2564
2565 let dir = tempfile::tempdir().expect("tempdir");
2566 let mut conn = temp_conn(&dir);
2567 apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("ledgered v1");
2568 conn.execute("INSERT INTO t1 (x) VALUES (7)", [])
2569 .expect("historical data");
2570
2571 let report =
2572 bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], Some(normalize_existing))
2573 .expect("bridge ledgered predecessor");
2574 assert_eq!(
2575 report,
2576 MaintenanceBridgeReport {
2577 from_version: 1,
2578 to_version: 2,
2579 prepared: 1,
2580 refused: Vec::new(),
2581 }
2582 );
2583 assert_eq!(
2584 domain_version(&conn, DOMAIN_V2.name).expect("ledger"),
2585 Some(2)
2586 );
2587 assert_eq!(
2588 conn.query_row("SELECT x FROM t1", [], |row| row.get::<_, i64>(0))
2589 .expect("prepared row"),
2590 8
2591 );
2592 conn.execute("INSERT INTO t1 (x, y) VALUES (9, 'migrated')", [])
2593 .expect("v2 shape");
2594 }
2595
2596 #[test]
2597 fn maintenance_bridge_prepares_existing_target_and_reports_durable_changes() {
2598 fn normalize_target(
2599 tx: &Transaction<'_>,
2600 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2601 let changed = tx.execute("UPDATE t1 SET x = x + 1", [])?;
2602 Ok(MaintenancePrepareReport::rewrote(changed))
2603 }
2604 fn mutate_then_fail(
2605 tx: &Transaction<'_>,
2606 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2607 tx.execute("UPDATE t1 SET x = 99", [])?;
2608 tx.execute_batch("THIS IS NOT SQL")?;
2609 Ok(MaintenancePrepareReport::rewrote(1))
2610 }
2611 fn rolls_back_then_begins_and_fails(
2612 tx: &Transaction<'_>,
2613 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2614 tx.execute_batch("UPDATE t1 SET x = 99; ROLLBACK; BEGIN; THIS IS NOT SQL")?;
2615 Ok(MaintenancePrepareReport::rewrote(1))
2616 }
2617
2618 let dir = tempfile::tempdir().expect("tempdir");
2619 let mut conn = temp_conn(&dir);
2620 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("ledgered target");
2621 conn.execute("INSERT INTO t1 (x, y) VALUES (7, 'target')", [])
2622 .expect("target data");
2623
2624 let report =
2625 bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], Some(normalize_target))
2626 .expect("prepare target");
2627 assert_eq!(
2628 report,
2629 MaintenanceBridgeReport {
2630 from_version: 2,
2631 to_version: 2,
2632 prepared: 1,
2633 refused: Vec::new(),
2634 }
2635 );
2636 assert!(!report.migrated());
2637 assert!(report.changed());
2638 assert_eq!(
2639 conn.query_row("SELECT x FROM t1", [], |row| row.get::<_, i64>(0))
2640 .expect("prepared target row"),
2641 8
2642 );
2643 assert_eq!(
2644 domain_version(&conn, DOMAIN_V2.name).expect("ledger"),
2645 Some(2)
2646 );
2647
2648 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], Some(mutate_then_fail))
2649 .expect_err("target prepare failure");
2650 assert!(matches!(
2651 err,
2652 SqliteStoreError::MigrationFailed { ref name, .. }
2653 if name == "maintenance-prepare"
2654 ));
2655 assert_eq!(
2656 conn.query_row("SELECT x FROM t1", [], |row| row.get::<_, i64>(0))
2657 .expect("row after rollback"),
2658 8
2659 );
2660
2661 let err = bridge_unledgered_domain(
2662 &mut conn,
2663 &DOMAIN_V2,
2664 2,
2665 &[1],
2666 Some(rolls_back_then_begins_and_fails),
2667 )
2668 .expect_err("target custody loss");
2669 assert!(matches!(
2670 err,
2671 SqliteStoreError::MigrationBrokeTransaction { ref name, .. }
2672 if name == "maintenance-prepare"
2673 ));
2674 assert_eq!(
2675 conn.query_row("SELECT x FROM t1", [], |row| row.get::<_, i64>(0))
2676 .expect("row after custody refusal"),
2677 8
2678 );
2679 assert_eq!(
2680 domain_version(&conn, DOMAIN_V2.name).expect("ledger"),
2681 Some(2)
2682 );
2683 }
2684
2685 #[test]
2686 fn maintenance_bridge_refuses_target_that_only_matches_migration_oracle() {
2687 let dir = tempfile::tempdir().expect("tempdir");
2688 let mut conn = temp_conn(&dir);
2689 conn.execute_batch("CREATE TABLE t1 (x INTEGER)")
2690 .expect("historical unledgered v1");
2691
2692 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2_ALT_INITIALIZER, 2, &[1], None)
2693 .expect_err("ordinary target verifier must reject drift");
2694 assert!(matches!(
2695 err,
2696 SqliteStoreError::SchemaFingerprintMismatch { version: 2, .. }
2697 ));
2698 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2699 let columns = conn
2700 .prepare("PRAGMA table_info(t1)")
2701 .expect("prepare")
2702 .query_map([], |row| row.get::<_, String>(1))
2703 .expect("query")
2704 .collect::<Result<Vec<_>, _>>()
2705 .expect("columns");
2706 assert_eq!(
2707 columns,
2708 vec!["x"],
2709 "target-verifier refusal did not roll back"
2710 );
2711 }
2712
2713 #[test]
2714 fn maintenance_bridge_refuses_catalog_outside_source_allowlist_across_data_only_gap() {
2715 fn no_op(_tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2716 Ok(())
2717 }
2718 const DATA_ONLY_GAP: SchemaDomain = SchemaDomain {
2719 name: "data-only-gap-domain",
2720 migrations: &[
2721 Migration {
2722 version: 1,
2723 name: "base",
2724 apply: create_t1,
2725 },
2726 Migration {
2727 version: 2,
2728 name: "data-only",
2729 apply: no_op,
2730 },
2731 Migration {
2732 version: 3,
2733 name: "add-y",
2734 apply: add_column_guarded,
2735 },
2736 ],
2737 initialize_current: initialize_v2,
2738 allowed_existing_versions: &[3],
2739 bridge_recoverable_versions: &[1],
2740 released_predecessors: &[],
2741 owned_objects: &[SchemaObject {
2742 kind: SchemaObjectKind::Table,
2743 name: "t1",
2744 }],
2745 retired_objects: &[],
2746 };
2747
2748 let dir = tempfile::tempdir().expect("tempdir");
2749 let mut conn = temp_conn(&dir);
2750 conn.execute_batch("CREATE TABLE t1 (x INTEGER)")
2751 .expect("v1-or-v2 catalog");
2752 let err = bridge_unledgered_domain(&mut conn, &DATA_ONLY_GAP, 3, &[3], None)
2753 .expect_err("excluded historical prefixes must not be inferred");
2754 assert!(matches!(
2755 err,
2756 SqliteStoreError::UnledgeredSchemaNoMatch { .. }
2757 ));
2758 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2759 let columns = conn
2760 .prepare("PRAGMA table_info(t1)")
2761 .expect("prepare")
2762 .query_map([], |row| row.get::<_, String>(1))
2763 .expect("query")
2764 .collect::<Result<Vec<_>, _>>()
2765 .expect("columns");
2766 assert_eq!(columns, vec!["x"]);
2767
2768 let err = bridge_unledgered_domain(&mut conn, &DATA_ONLY_GAP, 3, &[2, 1], None)
2769 .expect_err("unordered source authority must be refused");
2770 assert!(matches!(err, SqliteStoreError::InvalidMigrationList { .. }));
2771 }
2772
2773 #[test]
2774 fn maintenance_bridge_leaves_fresh_domain_for_normal_initializer() {
2775 let dir = tempfile::tempdir().expect("tempdir");
2776 let mut conn = temp_conn(&dir);
2777 let report =
2778 bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], None).expect("fresh no-op");
2779 assert_eq!(
2780 report,
2781 MaintenanceBridgeReport {
2782 from_version: 0,
2783 to_version: 0,
2784 prepared: 0,
2785 refused: Vec::new(),
2786 }
2787 );
2788 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2789 assert!(
2790 find_owned_objects(&conn, &DOMAIN_V2)
2791 .expect("objects")
2792 .is_empty()
2793 );
2794 }
2795
2796 #[test]
2797 fn maintenance_bridge_refuses_malformed_historical_shape_without_mutation() {
2798 let dir = tempfile::tempdir().expect("tempdir");
2799 let mut conn = temp_conn(&dir);
2800 conn.execute_batch("CREATE TABLE t1 (x INTEGER, candidate_only BLOB)")
2801 .expect("candidate schema");
2802
2803 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], None)
2804 .expect_err("refuse unauthenticated shape");
2805 assert!(matches!(
2806 err,
2807 SqliteStoreError::UnledgeredSchemaNoMatch {
2808 target_version: 2,
2809 ..
2810 }
2811 ));
2812 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2813 let columns = conn
2814 .prepare("PRAGMA table_info(t1)")
2815 .expect("prepare")
2816 .query_map([], |row| row.get::<_, String>(1))
2817 .expect("query")
2818 .collect::<Result<Vec<_>, _>>()
2819 .expect("columns");
2820 assert_eq!(columns, vec!["x", "candidate_only"]);
2821 }
2822
2823 #[test]
2824 fn maintenance_bridge_fingerprint_preserves_sql_literal_whitespace() {
2825 fn create_exact(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2826 tx.execute_batch("CREATE TABLE literal_t (x TEXT CHECK(x <> 'a b'))")
2827 }
2828 const LITERAL_DOMAIN: SchemaDomain = SchemaDomain {
2829 name: "literal-fingerprint-domain",
2830 migrations: &[Migration {
2831 version: 1,
2832 name: "base",
2833 apply: create_exact,
2834 }],
2835 initialize_current: create_exact,
2836 allowed_existing_versions: &[1],
2837 bridge_recoverable_versions: &[1],
2838 released_predecessors: &[],
2839 owned_objects: &[SchemaObject {
2840 kind: SchemaObjectKind::Table,
2841 name: "literal_t",
2842 }],
2843 retired_objects: &[],
2844 };
2845
2846 let dir = tempfile::tempdir().expect("tempdir");
2847 let mut conn = temp_conn(&dir);
2848 conn.execute_batch("CREATE TABLE literal_t (x TEXT CHECK(x <> 'a b'))")
2849 .expect("semantic mismatch");
2850 let err = bridge_unledgered_domain(&mut conn, &LITERAL_DOMAIN, 1, &[1], None)
2851 .expect_err("literal whitespace must remain fingerprint-significant");
2852 assert!(matches!(
2853 err,
2854 SqliteStoreError::UnledgeredSchemaNoMatch { .. }
2855 ));
2856 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2857
2858 assert_eq!(
2859 normalize_schema_sql("CREATE TABLE IF NOT EXISTS t (x CHECK(x <> 'IF NOT EXISTS'))"),
2860 "CREATE TABLE t (x CHECK(x <> 'IF NOT EXISTS'))"
2861 );
2862 }
2863
2864 #[test]
2865 fn maintenance_bridge_rolls_back_prepare_failure() {
2866 fn mutate_then_fail(
2867 tx: &Transaction<'_>,
2868 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2869 tx.execute("UPDATE t1 SET x = 99", [])?;
2870 tx.execute_batch("THIS IS NOT SQL")?;
2871 Ok(MaintenancePrepareReport::rewrote(1))
2872 }
2873
2874 let dir = tempfile::tempdir().expect("tempdir");
2875 let mut conn = temp_conn(&dir);
2876 conn.execute_batch("CREATE TABLE t1 (x INTEGER); INSERT INTO t1 VALUES (7)")
2877 .expect("historical unledgered v1");
2878
2879 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], Some(mutate_then_fail))
2880 .expect_err("prepare must fail");
2881 assert!(matches!(
2882 err,
2883 SqliteStoreError::MigrationFailed { ref name, .. }
2884 if name == "maintenance-prepare"
2885 ));
2886 assert_eq!(
2887 conn.query_row("SELECT x FROM t1", [], |row| row.get::<_, i64>(0))
2888 .expect("original row"),
2889 7
2890 );
2891 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2892 let columns = conn
2893 .prepare("PRAGMA table_info(t1)")
2894 .expect("prepare")
2895 .query_map([], |row| row.get::<_, String>(1))
2896 .expect("query")
2897 .collect::<Result<Vec<_>, _>>()
2898 .expect("columns");
2899 assert_eq!(columns, vec!["x"]);
2900 }
2901
2902 #[test]
2903 fn maintenance_bridge_reports_custody_loss_before_callback_error() {
2904 fn commits_then_fails(
2905 tx: &Transaction<'_>,
2906 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2907 tx.execute_batch("UPDATE t1 SET x = 99; COMMIT; THIS IS NOT SQL")?;
2908 Ok(MaintenancePrepareReport::rewrote(1))
2909 }
2910
2911 let dir = tempfile::tempdir().expect("tempdir");
2912 let mut conn = temp_conn(&dir);
2913 conn.execute_batch("CREATE TABLE t1 (x INTEGER); INSERT INTO t1 VALUES (7)")
2914 .expect("historical unledgered v1");
2915 let err =
2916 bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], Some(commits_then_fails))
2917 .expect_err("custody must dominate callback error");
2918 assert!(matches!(
2919 err,
2920 SqliteStoreError::MigrationBrokeTransaction { ref name, .. }
2921 if name == "maintenance-prepare"
2922 ));
2923 assert_eq!(domain_version(&conn, DOMAIN_V2.name).expect("ledger"), None);
2924 }
2925
2926 #[test]
2927 fn maintenance_bridge_refuses_undeclared_trigger_on_owned_table_before_prepare() {
2928 fn prepare_successor(
2929 tx: &Transaction<'_>,
2930 ) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2931 let changed = tx.execute("UPDATE t1 SET x = 8 WHERE x = 7", [])?;
2932 Ok(MaintenancePrepareReport::rewrote(changed))
2933 }
2934
2935 let dir = tempfile::tempdir().expect("tempdir");
2936 let mut conn = temp_conn(&dir);
2937 conn.execute_batch(
2938 "CREATE TABLE t1 (x INTEGER);
2939 INSERT INTO t1 VALUES (7);
2940 CREATE TRIGGER replace_prepared_successor
2941 AFTER UPDATE ON t1
2942 BEGIN
2943 UPDATE t1 SET x = 99 WHERE rowid = NEW.rowid;
2944 END",
2945 )
2946 .expect("intercepting trigger");
2947
2948 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], Some(prepare_successor))
2949 .expect_err("undeclared trigger must be refused before prepare");
2950 assert!(matches!(
2951 err,
2952 SqliteStoreError::SchemaFingerprintMismatch { version: 1, .. }
2953 ));
2954 assert_eq!(
2955 conn.query_row("SELECT x FROM t1", [], |row| row.get::<_, i64>(0))
2956 .expect("original row"),
2957 7
2958 );
2959 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
2960 }
2961
2962 #[test]
2963 fn maintenance_bridge_refuses_undeclared_instead_of_trigger_on_owned_view_before_prepare() {
2964 fn create_owned_view(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
2965 tx.execute_batch(
2966 "CREATE TABLE view_base (x INTEGER);
2967 CREATE VIEW owned_view AS SELECT x FROM view_base",
2968 )
2969 }
2970 fn prepare_view(tx: &Transaction<'_>) -> Result<MaintenancePrepareReport, rusqlite::Error> {
2971 let changed = tx.execute("UPDATE owned_view SET x = 8 WHERE x = 7", [])?;
2972 Ok(MaintenancePrepareReport::rewrote(changed))
2973 }
2974 const VIEW_DOMAIN: SchemaDomain = SchemaDomain {
2975 name: "view-bridge-domain",
2976 migrations: &[Migration {
2977 version: 1,
2978 name: "base",
2979 apply: create_owned_view,
2980 }],
2981 initialize_current: create_owned_view,
2982 allowed_existing_versions: &[1],
2983 bridge_recoverable_versions: &[1],
2984 released_predecessors: &[],
2985 owned_objects: &[
2986 SchemaObject {
2987 kind: SchemaObjectKind::Table,
2988 name: "view_base",
2989 },
2990 SchemaObject {
2991 kind: SchemaObjectKind::View,
2992 name: "owned_view",
2993 },
2994 ],
2995 retired_objects: &[],
2996 };
2997
2998 let dir = tempfile::tempdir().expect("tempdir");
2999 let mut conn = temp_conn(&dir);
3000 conn.execute_batch(
3001 "CREATE TABLE view_base (x INTEGER);
3002 CREATE VIEW owned_view AS SELECT x FROM view_base;
3003 INSERT INTO view_base VALUES (7);
3004 CREATE TRIGGER replace_view_update
3005 INSTEAD OF UPDATE ON owned_view
3006 BEGIN
3007 UPDATE view_base SET x = 99 WHERE x = OLD.x;
3008 END",
3009 )
3010 .expect("intercepting view trigger");
3011
3012 let err = bridge_unledgered_domain(&mut conn, &VIEW_DOMAIN, 1, &[1], Some(prepare_view))
3013 .expect_err("undeclared view trigger must be refused before prepare");
3014 assert!(matches!(
3015 err,
3016 SqliteStoreError::SchemaFingerprintMismatch { version: 1, .. }
3017 ));
3018 assert_eq!(
3019 conn.query_row("SELECT x FROM view_base", [], |row| row.get::<_, i64>(0))
3020 .expect("original row"),
3021 7
3022 );
3023 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
3024 }
3025
3026 #[test]
3027 fn maintenance_bridge_refuses_ambiguous_prefix_without_mutation() {
3028 fn no_op(_tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3029 Ok(())
3030 }
3031 const AMBIGUOUS: SchemaDomain = SchemaDomain {
3032 name: "ambiguous-bridge-domain",
3033 migrations: &[
3034 Migration {
3035 version: 1,
3036 name: "base",
3037 apply: create_t1,
3038 },
3039 Migration {
3040 version: 2,
3041 name: "data-only",
3042 apply: no_op,
3043 },
3044 ],
3045 initialize_current: create_t1,
3046 allowed_existing_versions: &[2],
3047 bridge_recoverable_versions: &[1],
3048 released_predecessors: &[],
3049 owned_objects: &[SchemaObject {
3050 kind: SchemaObjectKind::Table,
3051 name: "t1",
3052 }],
3053 retired_objects: &[],
3054 };
3055
3056 let dir = tempfile::tempdir().expect("tempdir");
3057 let mut conn = temp_conn(&dir);
3058 conn.execute_batch("CREATE TABLE t1 (x INTEGER)")
3059 .expect("ambiguous schema");
3060 let err = bridge_unledgered_domain(&mut conn, &AMBIGUOUS, 2, &[1, 2], None)
3061 .expect_err("refuse ambiguity");
3062 assert!(matches!(
3063 err,
3064 SqliteStoreError::UnledgeredSchemaAmbiguous { ref matches, .. }
3065 if matches == &[1, 2]
3066 ));
3067 assert!(!ledger_table_exists(&conn).expect("ledger presence"));
3068 }
3069
3070 #[test]
3071 fn maintenance_bridge_refuses_malformed_ledger_before_owned_schema_contact() {
3072 let dir = tempfile::tempdir().expect("tempdir");
3073 let mut conn = temp_conn(&dir);
3074 conn.execute_batch(
3075 "CREATE TABLE t1 (x INTEGER);
3076 CREATE TABLE meerkat_schema (domain TEXT, version INTEGER)",
3077 )
3078 .expect("malformed ledger");
3079 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], None)
3080 .expect_err("refuse malformed ledger");
3081 assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
3082 let columns = conn
3083 .prepare("PRAGMA table_info(t1)")
3084 .expect("prepare")
3085 .query_map([], |row| row.get::<_, String>(1))
3086 .expect("query")
3087 .collect::<Result<Vec<_>, _>>()
3088 .expect("columns");
3089 assert_eq!(columns, vec!["x"]);
3090 }
3091
3092 #[test]
3093 fn maintenance_bridge_refuses_mixed_case_trigger_that_mutates_foreign_ledger_row() {
3094 let dir = tempfile::tempdir().expect("tempdir");
3095 let mut conn = temp_conn(&dir);
3096 conn.execute_batch(
3097 "CREATE TABLE t1 (x INTEGER);
3098 CREATE TABLE meerkat_schema (
3099 domain TEXT PRIMARY KEY,
3100 version INTEGER NOT NULL
3101 );
3102 INSERT INTO meerkat_schema (domain, version) VALUES ('foreign-domain', 7);
3103 CREATE TRIGGER mutate_foreign_schema_row
3104 AFTER INSERT ON MEERKAT_SCHEMA
3105 BEGIN
3106 UPDATE MEERKAT_SCHEMA
3107 SET version = 999
3108 WHERE domain = 'foreign-domain';
3109 END",
3110 )
3111 .expect("hostile ledger trigger");
3112
3113 let err = bridge_unledgered_domain(&mut conn, &DOMAIN_V2, 2, &[1], None)
3114 .expect_err("mixed-case ledger trigger must be refused");
3115 assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
3116 let row_count = conn
3117 .query_row(
3118 "SELECT COUNT(*) FROM main.meerkat_schema WHERE domain = ?1",
3119 [DOMAIN_V2.name],
3120 |row| row.get::<_, i64>(0),
3121 )
3122 .expect("raw ledger count");
3123 assert_eq!(row_count, 0);
3124 let foreign_version = conn
3125 .query_row(
3126 "SELECT version FROM main.meerkat_schema WHERE domain = 'foreign-domain'",
3127 [],
3128 |row| row.get::<_, i64>(0),
3129 )
3130 .expect("foreign ledger row");
3131 assert_eq!(foreign_version, 7);
3132 let columns = conn
3133 .prepare("PRAGMA table_info(t1)")
3134 .expect("prepare")
3135 .query_map([], |row| row.get::<_, String>(1))
3136 .expect("query")
3137 .collect::<Result<Vec<_>, _>>()
3138 .expect("columns");
3139 assert_eq!(
3140 columns,
3141 vec!["x"],
3142 "failed stamp did not roll back migration"
3143 );
3144 }
3145
3146 #[test]
3147 fn fresh_domain_ignores_foreign_cotenant_objects() {
3148 let dir = tempfile::tempdir().expect("tempdir");
3149 let mut conn = temp_conn(&dir);
3150 conn.execute_batch("CREATE TABLE foreign_table (value TEXT)")
3151 .expect("foreign ddl");
3152 let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("fresh domain");
3153 assert_eq!(
3154 report,
3155 LedgerReport {
3156 from_version: 0,
3157 to_version: 2
3158 }
3159 );
3160 conn.execute("INSERT INTO foreign_table VALUES ('kept')", [])
3161 .expect("foreign object survives");
3162 }
3163
3164 #[test]
3165 fn future_version_is_refused_before_any_mutation() {
3166 let dir = tempfile::tempdir().expect("tempdir");
3167 let mut conn = temp_conn(&dir);
3168 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("stamp v2");
3169 let err = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect_err("refuse");
3171 match err {
3172 SqliteStoreError::SchemaFromTheFuture {
3173 domain,
3174 found,
3175 supported,
3176 } => {
3177 assert_eq!(domain, "test-domain");
3178 assert_eq!(found, 2);
3179 assert_eq!(supported, 1);
3180 }
3181 other => panic!("wrong error: {other}"),
3182 }
3183 assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(2));
3185 }
3186
3187 #[test]
3188 fn foreign_domain_rows_are_untouched() {
3189 let dir = tempfile::tempdir().expect("tempdir");
3190 let mut conn = temp_conn(&dir);
3191 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("mine");
3192 conn.execute(
3193 "INSERT INTO meerkat_schema (domain, version) VALUES ('foreign-domain', 7)",
3194 [],
3195 )
3196 .expect("foreign row");
3197 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("noop");
3198 let foreign: i64 = conn
3199 .query_row(
3200 "SELECT version FROM meerkat_schema WHERE domain = 'foreign-domain'",
3201 [],
3202 |r| r.get(0),
3203 )
3204 .expect("foreign row survives");
3205 assert_eq!(foreign, 7);
3206 }
3207
3208 #[test]
3209 fn invalid_migration_list_is_refused_without_touching_the_file() {
3210 const BAD: SchemaDomain = SchemaDomain {
3211 name: "bad-domain",
3212 migrations: &[Migration {
3213 version: 3,
3214 name: "gap",
3215 apply: create_t1,
3216 }],
3217 initialize_current: create_t1,
3218 allowed_existing_versions: &[3],
3219 bridge_recoverable_versions: &[1],
3220 released_predecessors: &[],
3221 owned_objects: &[],
3222 retired_objects: &[],
3223 };
3224 let dir = tempfile::tempdir().expect("tempdir");
3225 let mut conn = temp_conn(&dir);
3226 let err = apply_domain_migrations(&mut conn, &BAD).expect_err("refuse");
3227 assert!(matches!(err, SqliteStoreError::InvalidMigrationList { .. }));
3228 assert!(!ledger_table_exists(&conn).expect("check"));
3229 }
3230
3231 #[test]
3232 fn failed_migration_rolls_back_atomically() {
3233 fn fail(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3234 tx.execute_batch("CREATE TABLE half_done (x INTEGER)")?;
3235 tx.execute_batch("THIS IS NOT SQL")
3236 }
3237 fn initialize_failing(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3238 create_t1(tx)?;
3239 fail(tx)
3240 }
3241 const FAILING: SchemaDomain = SchemaDomain {
3242 name: "failing-domain",
3243 migrations: &[
3244 Migration {
3245 version: 1,
3246 name: "base",
3247 apply: create_t1,
3248 },
3249 Migration {
3250 version: 2,
3251 name: "explodes",
3252 apply: fail,
3253 },
3254 ],
3255 initialize_current: initialize_failing,
3256 allowed_existing_versions: &[1, 2],
3257 bridge_recoverable_versions: &[1],
3258 released_predecessors: &[SchemaPredecessor {
3259 version: 1,
3260 verify: verify_v1,
3261 }],
3262 owned_objects: &[
3263 SchemaObject {
3264 kind: SchemaObjectKind::Table,
3265 name: "t1",
3266 },
3267 SchemaObject {
3268 kind: SchemaObjectKind::Table,
3269 name: "half_done",
3270 },
3271 ],
3272 retired_objects: &[],
3273 };
3274 let dir = tempfile::tempdir().expect("tempdir");
3275 let mut conn = temp_conn(&dir);
3276 let err = apply_domain_migrations(&mut conn, &FAILING).expect_err("must fail");
3277 assert!(matches!(
3278 err,
3279 SqliteStoreError::MigrationFailed { version: 2, .. }
3280 ));
3281 assert_eq!(domain_version(&conn, "failing-domain").expect("read"), None);
3284 let tables: Vec<String> = conn
3285 .prepare(
3286 "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('t1','half_done')",
3287 )
3288 .expect("prepare")
3289 .query_map([], |r| r.get(0))
3290 .expect("query")
3291 .collect::<Result<_, _>>()
3292 .expect("rows");
3293 assert!(tables.is_empty(), "rollback left tables behind: {tables:?}");
3294 }
3295
3296 #[test]
3297 fn malformed_ledger_shape_is_refused_not_healed() {
3298 let dir = tempfile::tempdir().expect("tempdir");
3299 let mut conn = temp_conn(&dir);
3300 conn.execute_batch("CREATE TABLE meerkat_schema (x INTEGER)")
3302 .expect("foreign ddl");
3303 let err = domain_version(&conn, "test-domain").expect_err("refuse read");
3304 assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
3305 let err = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect_err("refuse migrate");
3306 assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
3307 let count: i64 = conn
3309 .query_row("SELECT COUNT(*) FROM meerkat_schema", [], |r| r.get(0))
3310 .expect("foreign table survives");
3311 assert_eq!(count, 0);
3312 }
3313
3314 #[test]
3315 fn non_positive_versions_are_refused_not_healed() {
3316 for bad_version in [0i64, -3] {
3317 let dir = tempfile::tempdir().expect("tempdir");
3318 let mut conn = temp_conn(&dir);
3319 conn.execute_batch(
3320 "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL)",
3321 )
3322 .expect("ledger ddl");
3323 conn.execute(
3324 "INSERT INTO meerkat_schema (domain, version) VALUES ('test-domain', ?1)",
3325 [bad_version],
3326 )
3327 .expect("seed bad version");
3328 let err = domain_version(&conn, "test-domain").expect_err("refuse read");
3329 assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
3330 let err = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect_err("refuse migrate");
3331 assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
3332 let stored: i64 = conn
3334 .query_row(
3335 "SELECT version FROM meerkat_schema WHERE domain = 'test-domain'",
3336 [],
3337 |r| r.get(0),
3338 )
3339 .expect("row survives");
3340 assert_eq!(stored, bad_version);
3341 }
3342 }
3343
3344 #[test]
3345 fn duplicate_domain_rows_are_refused() {
3346 let dir = tempfile::tempdir().expect("tempdir");
3347 let conn = temp_conn(&dir);
3348 conn.execute_batch(
3352 "CREATE TABLE meerkat_schema (domain TEXT, version INTEGER NOT NULL);
3353 INSERT INTO meerkat_schema VALUES ('dup-domain', 1);
3354 INSERT INTO meerkat_schema VALUES ('dup-domain', 2);",
3355 )
3356 .expect("seed duplicates");
3357 let err = read_version(&conn, "dup-domain").expect_err("refuse duplicates");
3358 match err {
3359 SqliteStoreError::LedgerMalformed { detail } => {
3360 assert!(detail.contains("multiple ledger rows"), "{detail}");
3361 }
3362 other => panic!("wrong error: {other}"),
3363 }
3364 }
3365
3366 #[test]
3367 fn temp_shadowing_cannot_hijack_the_ledger() {
3368 let dir = tempfile::tempdir().expect("tempdir");
3369 let mut conn = temp_conn(&dir);
3370 apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("stamp v1");
3371 conn.execute_batch(
3374 "CREATE TEMP TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
3375 INSERT INTO temp.meerkat_schema VALUES ('test-domain', 999);",
3376 )
3377 .expect("temp shadow");
3378 assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(1));
3379 let report = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("noop against main");
3380 assert!(!report.migrated());
3381 }
3382
3383 #[test]
3384 fn migration_that_ends_the_transaction_is_refused_unstamped() {
3385 fn no_op(_tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3386 Ok(())
3387 }
3388 fn verify_empty_predecessor(_conn: &Connection) -> Result<(), String> {
3389 Ok(())
3390 }
3391 fn commits_underneath(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3392 tx.execute_batch("CREATE TABLE escaped_commit (x INTEGER); COMMIT")
3393 }
3394 fn rolls_back_underneath(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3395 tx.execute_batch("ROLLBACK")
3396 }
3397 fn commits_then_begins(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3401 tx.execute_batch("CREATE TABLE escaped_commit_begin (x INTEGER); COMMIT; BEGIN")
3402 }
3403 fn rolls_back_then_begins(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3404 tx.execute_batch("ROLLBACK; BEGIN")
3405 }
3406 const COMMITS: SchemaDomain = SchemaDomain {
3407 name: "custody-commit",
3408 migrations: &[
3409 Migration {
3410 version: 1,
3411 name: "base",
3412 apply: no_op,
3413 },
3414 Migration {
3415 version: 2,
3416 name: "commits-underneath",
3417 apply: commits_underneath,
3418 },
3419 ],
3420 initialize_current: no_op,
3421 allowed_existing_versions: &[1, 2],
3422 bridge_recoverable_versions: &[1],
3423 released_predecessors: &[SchemaPredecessor {
3424 version: 1,
3425 verify: verify_empty_predecessor,
3426 }],
3427 owned_objects: &[],
3428 retired_objects: &[],
3429 };
3430 const ROLLS_BACK: SchemaDomain = SchemaDomain {
3431 name: "custody-rollback",
3432 migrations: &[
3433 Migration {
3434 version: 1,
3435 name: "base",
3436 apply: no_op,
3437 },
3438 Migration {
3439 version: 2,
3440 name: "rolls-back-underneath",
3441 apply: rolls_back_underneath,
3442 },
3443 ],
3444 initialize_current: no_op,
3445 allowed_existing_versions: &[1, 2],
3446 bridge_recoverable_versions: &[1],
3447 released_predecessors: &[SchemaPredecessor {
3448 version: 1,
3449 verify: verify_empty_predecessor,
3450 }],
3451 owned_objects: &[],
3452 retired_objects: &[],
3453 };
3454 const COMMITS_THEN_BEGINS: SchemaDomain = SchemaDomain {
3455 name: "custody-commit-begin",
3456 migrations: &[
3457 Migration {
3458 version: 1,
3459 name: "base",
3460 apply: no_op,
3461 },
3462 Migration {
3463 version: 2,
3464 name: "commits-then-begins",
3465 apply: commits_then_begins,
3466 },
3467 ],
3468 initialize_current: no_op,
3469 allowed_existing_versions: &[1, 2],
3470 bridge_recoverable_versions: &[1],
3471 released_predecessors: &[SchemaPredecessor {
3472 version: 1,
3473 verify: verify_empty_predecessor,
3474 }],
3475 owned_objects: &[],
3476 retired_objects: &[],
3477 };
3478 const ROLLS_BACK_THEN_BEGINS: SchemaDomain = SchemaDomain {
3479 name: "custody-rollback-begin",
3480 migrations: &[
3481 Migration {
3482 version: 1,
3483 name: "base",
3484 apply: no_op,
3485 },
3486 Migration {
3487 version: 2,
3488 name: "rolls-back-then-begins",
3489 apply: rolls_back_then_begins,
3490 },
3491 ],
3492 initialize_current: no_op,
3493 allowed_existing_versions: &[1, 2],
3494 bridge_recoverable_versions: &[1],
3495 released_predecessors: &[SchemaPredecessor {
3496 version: 1,
3497 verify: verify_empty_predecessor,
3498 }],
3499 owned_objects: &[],
3500 retired_objects: &[],
3501 };
3502 for (domain, expected_name) in [
3503 (&COMMITS, "commits-underneath"),
3504 (&ROLLS_BACK, "rolls-back-underneath"),
3505 (&COMMITS_THEN_BEGINS, "commits-then-begins"),
3506 (&ROLLS_BACK_THEN_BEGINS, "rolls-back-then-begins"),
3507 ] {
3508 let dir = tempfile::tempdir().expect("tempdir");
3509 let mut conn = temp_conn(&dir);
3510 conn.execute_batch(CREATE_LEDGER_SQL).expect("ledger");
3511 conn.execute(
3512 "INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, 1)",
3513 [domain.name],
3514 )
3515 .expect("released predecessor");
3516 let err = apply_domain_migrations(&mut conn, domain).expect_err("custody violation");
3517 match err {
3518 SqliteStoreError::MigrationBrokeTransaction {
3519 domain: err_domain,
3520 version,
3521 name,
3522 } => {
3523 assert_eq!(err_domain, domain.name);
3524 assert_eq!(version, 2);
3525 assert_eq!(name, expected_name);
3526 }
3527 other => panic!("wrong error: {other}"),
3528 }
3529 assert_eq!(domain_version(&conn, domain.name).expect("read"), Some(1));
3532 }
3533 }
3534
3535 #[test]
3536 fn initializer_that_ends_the_transaction_is_refused_unstamped() {
3537 fn commits_underneath(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3538 tx.execute_batch("CREATE TABLE escaped_initializer (x INTEGER); COMMIT")
3539 }
3540 const COMMITS: SchemaDomain = SchemaDomain {
3541 name: "initializer-custody-commit",
3542 migrations: &[Migration {
3543 version: 1,
3544 name: "base",
3545 apply: commits_underneath,
3546 }],
3547 initialize_current: commits_underneath,
3548 allowed_existing_versions: &[1],
3549 bridge_recoverable_versions: &[1],
3550 released_predecessors: &[],
3551 owned_objects: &[SchemaObject {
3552 kind: SchemaObjectKind::Table,
3553 name: "escaped_initializer",
3554 }],
3555 retired_objects: &[],
3556 };
3557 let dir = tempfile::tempdir().expect("tempdir");
3558 let mut conn = temp_conn(&dir);
3559 let err = apply_domain_migrations(&mut conn, &COMMITS).expect_err("custody violation");
3560 match err {
3561 SqliteStoreError::MigrationBrokeTransaction {
3562 domain,
3563 version,
3564 name,
3565 } => {
3566 assert_eq!(domain, COMMITS.name);
3567 assert_eq!(version, 1);
3568 assert_eq!(name, "initialize-current");
3569 }
3570 other => panic!("wrong error: {other}"),
3571 }
3572 assert_eq!(domain_version(&conn, COMMITS.name).expect("read"), None);
3573 }
3574
3575 #[test]
3576 fn initializer_using_its_own_savepoints_keeps_custody() {
3577 fn nests_savepoints(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
3581 tx.execute_batch(
3582 "SAVEPOINT body_sp;
3583 CREATE TABLE sp_t (x INTEGER);
3584 RELEASE SAVEPOINT body_sp",
3585 )
3586 }
3587 const NESTED: SchemaDomain = SchemaDomain {
3588 name: "custody-nested-savepoint",
3589 migrations: &[Migration {
3590 version: 1,
3591 name: "nests-savepoints",
3592 apply: nests_savepoints,
3593 }],
3594 initialize_current: nests_savepoints,
3595 allowed_existing_versions: &[1],
3596 bridge_recoverable_versions: &[1],
3597 released_predecessors: &[],
3598 owned_objects: &[SchemaObject {
3599 kind: SchemaObjectKind::Table,
3600 name: "sp_t",
3601 }],
3602 retired_objects: &[],
3603 };
3604 let dir = tempfile::tempdir().expect("tempdir");
3605 let mut conn = temp_conn(&dir);
3606 let report = apply_domain_migrations(&mut conn, &NESTED).expect("apply");
3607 assert_eq!(report.to_version, 1);
3608 assert_eq!(domain_version(&conn, NESTED.name).expect("read"), Some(1));
3609 }
3610
3611 #[test]
3612 fn schema_preflight_passes_fresh_and_current_refuses_future() {
3613 let dir = tempfile::tempdir().expect("tempdir");
3614 let mut conn = temp_conn(&dir);
3615 preflight_schema_eligibility(&conn, &DOMAIN_V1).expect("no ledger yet");
3616 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("stamp v2");
3617 preflight_schema_eligibility(&conn, &DOMAIN_V2).expect("current");
3618 let err =
3619 preflight_schema_eligibility(&conn, &DOMAIN_V1).expect_err("future for old binary");
3620 assert!(matches!(
3621 err,
3622 SqliteStoreError::SchemaFromTheFuture {
3623 found: 2,
3624 supported: 1,
3625 ..
3626 }
3627 ));
3628 }
3629
3630 #[test]
3631 fn current_schema_verification_does_not_take_the_wal_writer() {
3632 let dir = tempfile::tempdir().expect("tempdir");
3633 let path = dir.path().join("db.sqlite3");
3634 let mut writer = open(&path, ConnectionProfile::PRIMARY).expect("writer");
3635 apply_domain_migrations(&mut writer, &DOMAIN_V2).expect("initialize");
3636 let mut reader = open(&path, ConnectionProfile::PRIMARY).expect("reader");
3637 reader
3640 .busy_timeout(std::time::Duration::ZERO)
3641 .expect("busy policy");
3642 let tx = writer
3643 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
3644 .expect("reserve unrelated writer");
3645 let report = apply_domain_migrations(&mut reader, &DOMAIN_V2)
3646 .expect("current schema verification must be a read-only snapshot");
3647 assert!(!report.migrated());
3648 assert!(reader.is_autocommit());
3649 drop(tx);
3650 }
3651
3652 #[test]
3653 fn concurrent_opens_race_safely() {
3654 let dir = tempfile::tempdir().expect("tempdir");
3655 let path = dir.path().join("db.sqlite3");
3656 let mut handles = Vec::new();
3657 for _ in 0..8 {
3658 let path = path.clone();
3659 handles.push(std::thread::spawn(move || {
3660 let mut conn = open(&path, ConnectionProfile::PRIMARY).expect("open");
3661 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("apply")
3662 }));
3663 }
3664 let mut migrated = 0;
3665 for handle in handles {
3666 let report = handle.join().expect("thread");
3667 assert_eq!(report.to_version, 2);
3668 if report.migrated() {
3669 migrated += 1;
3670 }
3671 }
3672 assert_eq!(migrated, 1, "only the winning writer may initialize");
3673 let conn = open(&path, ConnectionProfile::ReadOnly).expect("reopen");
3674 assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(2));
3675 }
3676
3677 #[test]
3678 fn concurrent_predecessor_upgrade_rechecks_after_read_snapshot() {
3679 let dir = tempfile::tempdir().expect("tempdir");
3680 let path = dir.path().join("db.sqlite3");
3681 let mut conn = open(&path, ConnectionProfile::PRIMARY).expect("seed");
3682 apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("version 1");
3683 let barrier = std::sync::Arc::new(std::sync::Barrier::new(8));
3684 let handles = (0..8)
3685 .map(|_| {
3686 let path = path.clone();
3687 let barrier = barrier.clone();
3688 std::thread::spawn(move || {
3689 let mut conn = open(&path, ConnectionProfile::PRIMARY).expect("open");
3690 barrier.wait();
3691 apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("upgrade")
3692 })
3693 })
3694 .collect::<Vec<_>>();
3695 let reports = handles
3696 .into_iter()
3697 .map(|thread| thread.join().expect("thread"))
3698 .collect::<Vec<_>>();
3699 assert_eq!(reports.iter().filter(|report| report.migrated()).count(), 1);
3700 assert!(reports.iter().all(|report| report.to_version == 2));
3701 preflight_schema_eligibility(&conn, &DOMAIN_V2).expect("verified upgraded catalog");
3702 }
3703}