Skip to main content

meerkat_sqlite/
ledger.rs

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