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