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