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