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