use rusqlite::{Connection, OptionalExtension, Transaction};
use std::collections::BTreeMap;
use std::sync::{Mutex, OnceLock};
use crate::error::SqliteStoreError;
const CREATE_LEDGER_SQL: &str = "CREATE TABLE IF NOT EXISTS main.meerkat_schema (
domain TEXT PRIMARY KEY,
version INTEGER NOT NULL
)";
const CUSTODY_SAVEPOINT_SQL: &str = "SAVEPOINT meerkat_migration_custody";
const CUSTODY_RELEASE_SQL: &str = "RELEASE SAVEPOINT meerkat_migration_custody";
#[derive(Debug)]
pub struct Migration {
pub version: i64,
pub name: &'static str,
pub apply: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
}
#[derive(Debug)]
pub struct SchemaPredecessor {
pub version: i64,
pub verify: fn(&Connection) -> Result<(), String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SchemaObjectKind {
Table,
Index,
Trigger,
View,
}
impl SchemaObjectKind {
fn sqlite_name(self) -> &'static str {
match self {
Self::Table => "table",
Self::Index => "index",
Self::Trigger => "trigger",
Self::View => "view",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SchemaObject {
pub kind: SchemaObjectKind,
pub name: &'static str,
}
#[derive(Debug)]
pub struct SchemaDomain {
pub name: &'static str,
pub migrations: &'static [Migration],
pub initialize_current: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
pub allowed_existing_versions: &'static [i64],
pub released_predecessors: &'static [SchemaPredecessor],
pub owned_objects: &'static [SchemaObject],
pub retired_objects: &'static [SchemaObject],
}
impl SchemaDomain {
pub fn supported_version(&self) -> i64 {
self.migrations.last().map_or(0, |m| m.version)
}
fn validate(&self) -> Result<(), SqliteStoreError> {
for (idx, migration) in self.migrations.iter().enumerate() {
let expected = idx as i64 + 1;
if migration.version != expected {
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!(
"migration at position {idx} has version {}, expected {expected} \
(versions must be contiguous from 1)",
migration.version
),
});
}
}
let supported = self.supported_version();
let mut previous = None;
for &version in self.allowed_existing_versions {
if version <= 0 || version > supported {
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!(
"allowed existing version {version} is outside 1..={supported}"
),
});
}
if previous.is_some_and(|value| value >= version) {
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: "allowed existing versions must be strictly increasing".to_string(),
});
}
previous = Some(version);
}
if !self.allowed_existing_versions.contains(&supported) {
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!(
"allowed existing versions must explicitly include current version {supported}"
),
});
}
for &version in self
.allowed_existing_versions
.iter()
.filter(|&&version| version < supported)
{
let matches = self
.released_predecessors
.iter()
.filter(|predecessor| predecessor.version == version)
.count();
if matches != 1 {
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!(
"allowed predecessor version {version} must have exactly one frozen \
verifier, found {matches}"
),
});
}
}
for predecessor in self.released_predecessors {
if predecessor.version >= supported
|| !self
.allowed_existing_versions
.contains(&predecessor.version)
{
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!(
"fingerprint verifier for version {} is not an allowed predecessor",
predecessor.version
),
});
}
}
for (idx, object) in self
.owned_objects
.iter()
.chain(self.retired_objects)
.enumerate()
{
if object.name.is_empty() || object.name == "meerkat_schema" {
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!(
"owned object at position {idx} has reserved or empty name `{}`",
object.name
),
});
}
if self
.owned_objects
.iter()
.chain(self.retired_objects)
.take(idx)
.any(|prior| prior.name == object.name)
{
return Err(SqliteStoreError::InvalidMigrationList {
domain: self.name.to_string(),
detail: format!("owned object name `{}` is duplicated", object.name),
});
}
}
Ok(())
}
fn accepts_existing_version(&self, version: i64) -> bool {
self.allowed_existing_versions.contains(&version)
}
fn verify_predecessor(&self, conn: &Connection, version: i64) -> Result<(), SqliteStoreError> {
if version == self.supported_version() {
return verify_current_schema_fingerprint(conn, self).map_err(|detail| {
SqliteStoreError::SchemaFingerprintMismatch {
domain: self.name.to_string(),
version,
detail,
}
});
}
let predecessor = self
.released_predecessors
.iter()
.find(|predecessor| predecessor.version == version)
.ok_or_else(|| unsupported_predecessor(self, version))?;
(predecessor.verify)(conn).map_err(|detail| SqliteStoreError::SchemaFingerprintMismatch {
domain: self.name.to_string(),
version,
detail,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LedgerReport {
pub from_version: i64,
pub to_version: i64,
}
impl LedgerReport {
pub fn migrated(&self) -> bool {
self.to_version > self.from_version
}
}
pub fn domain_version(conn: &Connection, domain: &str) -> Result<Option<i64>, SqliteStoreError> {
if !ledger_table_exists(conn)? {
return Ok(None);
}
validate_ledger_shape(conn)?;
read_version(conn, domain)
}
pub fn preflight_schema_eligibility(
conn: &Connection,
domain: &SchemaDomain,
) -> Result<(), SqliteStoreError> {
domain.validate()?;
let supported = domain.supported_version();
match domain_version(conn, domain.name)? {
Some(found) if found > supported => {
return Err(SqliteStoreError::SchemaFromTheFuture {
domain: domain.name.to_string(),
found,
supported,
});
}
Some(found) if !domain.accepts_existing_version(found) => {
return Err(unsupported_predecessor(domain, found));
}
Some(found) => domain.verify_predecessor(conn, found)?,
None => {
let objects = find_owned_objects(conn, domain)?;
if !objects.is_empty() {
return Err(SqliteStoreError::UnledgeredDomainObjects {
domain: domain.name.to_string(),
objects,
});
}
}
}
Ok(())
}
pub fn apply_domain_migrations(
conn: &mut Connection,
domain: &SchemaDomain,
) -> Result<LedgerReport, SqliteStoreError> {
domain.validate()?;
let supported = domain.supported_version();
let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
let current = if ledger_table_exists(&tx)? {
validate_ledger_shape(&tx)?;
read_version(&tx, domain.name)?
} else {
None
};
if let Some(found) = current {
if found > supported {
return Err(SqliteStoreError::SchemaFromTheFuture {
domain: domain.name.to_string(),
found,
supported,
});
}
if !domain.accepts_existing_version(found) {
return Err(unsupported_predecessor(domain, found));
}
domain.verify_predecessor(&tx, found)?;
} else {
let objects = find_owned_objects(&tx, domain)?;
if !objects.is_empty() {
return Err(SqliteStoreError::UnledgeredDomainObjects {
domain: domain.name.to_string(),
objects,
});
}
}
let current = current.unwrap_or(0);
if current == supported {
return Ok(LedgerReport {
from_version: current,
to_version: current,
});
}
if !ledger_table_exists(&tx)? {
tx.execute_batch(CREATE_LEDGER_SQL)?;
validate_ledger_shape(&tx)?;
}
if current == 0 {
tx.execute_batch(CUSTODY_SAVEPOINT_SQL)?;
(domain.initialize_current)(&tx).map_err(|source| SqliteStoreError::MigrationFailed {
domain: domain.name.to_string(),
version: supported,
name: "initialize-current".to_string(),
source,
})?;
if tx.is_autocommit() || tx.execute_batch(CUSTODY_RELEASE_SQL).is_err() {
return Err(SqliteStoreError::MigrationBrokeTransaction {
domain: domain.name.to_string(),
version: supported,
name: "initialize-current".to_string(),
});
}
} else {
for migration in domain.migrations.iter().filter(|m| m.version > current) {
tx.execute_batch(CUSTODY_SAVEPOINT_SQL)?;
(migration.apply)(&tx).map_err(|source| SqliteStoreError::MigrationFailed {
domain: domain.name.to_string(),
version: migration.version,
name: migration.name.to_string(),
source,
})?;
if tx.is_autocommit() || tx.execute_batch(CUSTODY_RELEASE_SQL).is_err() {
return Err(SqliteStoreError::MigrationBrokeTransaction {
domain: domain.name.to_string(),
version: migration.version,
name: migration.name.to_string(),
});
}
}
}
verify_current_schema_fingerprint(&tx, domain).map_err(|detail| {
SqliteStoreError::SchemaFingerprintMismatch {
domain: domain.name.to_string(),
version: supported,
detail,
}
})?;
tx.execute(
"INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, ?2)
ON CONFLICT(domain) DO UPDATE SET version = excluded.version",
rusqlite::params![domain.name, supported],
)?;
tx.commit()?;
Ok(LedgerReport {
from_version: current,
to_version: supported,
})
}
static EXPECTED_CURRENT_CATALOGS: OnceLock<Mutex<BTreeMap<String, Result<String, String>>>> =
OnceLock::new();
fn verify_current_schema_fingerprint(
actual: &Connection,
domain: &SchemaDomain,
) -> Result<(), String> {
let expected = {
let cache = EXPECTED_CURRENT_CATALOGS.get_or_init(|| Mutex::new(BTreeMap::new()));
let key = current_catalog_cache_key(domain);
let cached = cache
.lock()
.map_err(|_| "current catalog cache lock is poisoned".to_string())?
.get(&key)
.cloned();
if let Some(cached) = cached {
cached?
} else {
let built = build_current_catalog_fingerprint(domain);
cache
.lock()
.map_err(|_| "current catalog cache lock is poisoned".to_string())?
.insert(key, built.clone());
built?
}
};
let actual = compact_catalog_fingerprint(actual, domain, domain.owned_objects)?;
if actual != expected {
return Err(format!(
"current owned catalog differs: expected {expected}, found {actual}"
));
}
Ok(())
}
fn current_catalog_cache_key(domain: &SchemaDomain) -> String {
let mut key = format!(
"{}\u{1f}{}\u{1f}{:x}",
domain.name,
domain.supported_version(),
domain.initialize_current as usize
);
for object in domain.owned_objects {
key.push_str(&format!(
"\u{1e}current:{}:{}",
object.kind.sqlite_name(),
object.name
));
}
for object in domain.retired_objects {
key.push_str(&format!(
"\u{1e}retired:{}:{}",
object.kind.sqlite_name(),
object.name
));
}
key
}
fn build_current_catalog_fingerprint(domain: &SchemaDomain) -> Result<String, String> {
let mut expected =
Connection::open_in_memory().map_err(|error| format!("open current oracle: {error}"))?;
let tx = expected
.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
.map_err(|error| format!("begin current oracle: {error}"))?;
(domain.initialize_current)(&tx).map_err(|error| format!("build current oracle: {error}"))?;
tx.commit()
.map_err(|error| format!("commit current oracle: {error}"))?;
compact_catalog_fingerprint(&expected, domain, domain.owned_objects)
}
fn compact_catalog_fingerprint(
conn: &Connection,
domain: &SchemaDomain,
expected_objects: &[SchemaObject],
) -> Result<String, String> {
let all_objects = all_domain_objects(domain);
let owned_by_name = all_objects
.iter()
.map(|object| (object.name, object))
.collect::<BTreeMap<_, _>>();
let current_by_name = expected_objects
.iter()
.map(|object| (object.name, object))
.collect::<BTreeMap<_, _>>();
let mut actual_names = Vec::new();
let mut entries = Vec::with_capacity(expected_objects.len());
let mut statement = conn
.prepare(
"SELECT type, name, tbl_name, sql
FROM main.sqlite_schema
WHERE name NOT LIKE 'sqlite_%'
ORDER BY type, name",
)
.map_err(|error| error.to_string())?;
let rows = statement
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
))
})
.map_err(|error| error.to_string())?;
for row in rows {
let (kind, name, table_name, sql) = row.map_err(|error| error.to_string())?;
if owned_by_name.contains_key(name.as_str()) {
actual_names.push((kind.clone(), name.clone()));
}
if current_by_name.contains_key(name.as_str()) {
entries.push(format!(
"{kind}\u{1f}{name}\u{1f}{table_name}\u{1f}{}",
sql.map(|sql| normalize_schema_sql(&sql))
.unwrap_or_default()
));
}
}
actual_names.sort();
let mut expected_names = expected_objects
.iter()
.map(|object| {
(
object.kind.sqlite_name().to_string(),
object.name.to_string(),
)
})
.collect::<Vec<_>>();
expected_names.sort();
if actual_names != expected_names {
return Err(format!(
"owned object set differs: expected {expected_names:?}, found {actual_names:?}"
));
}
entries.sort();
Ok(entries.join("\u{1e}"))
}
fn all_domain_objects(domain: &SchemaDomain) -> Vec<SchemaObject> {
domain
.owned_objects
.iter()
.chain(domain.retired_objects)
.copied()
.collect()
}
pub fn verify_released_schema_fingerprint(
actual: &Connection,
domain: &SchemaDomain,
released_objects: &[SchemaObject],
build_released: fn(&Transaction<'_>) -> Result<(), rusqlite::Error>,
) -> Result<(), String> {
let mut expected = Connection::open_in_memory()
.map_err(|error| format!("open fingerprint oracle: {error}"))?;
let tx = expected
.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
.map_err(|error| format!("begin fingerprint oracle: {error}"))?;
build_released(&tx).map_err(|error| format!("build fingerprint oracle: {error}"))?;
tx.commit()
.map_err(|error| format!("commit fingerprint oracle: {error}"))?;
let expected_names = catalog_names(&expected, released_objects)
.map_err(|error| format!("read fingerprint oracle: {error}"))?;
let mut declared_expected = released_objects
.iter()
.map(|object| {
(
object.kind.sqlite_name().to_string(),
object.name.to_string(),
)
})
.collect::<Vec<_>>();
declared_expected.sort();
if expected_names != declared_expected {
return Err(format!(
"frozen builder produced {expected_names:?}, manifest declares {declared_expected:?}"
));
}
let actual_names = catalog_names(actual, &all_domain_objects(domain))
.map_err(|error| format!("read actual catalog: {error}"))?;
if actual_names != declared_expected {
return Err(format!(
"owned object set differs: expected {declared_expected:?}, found {actual_names:?}"
));
}
for object in released_objects {
let wanted = catalog_fingerprint(&expected, object)
.map_err(|error| format!("fingerprint oracle {}: {error}", object.name))?;
let found = catalog_fingerprint(actual, object)
.map_err(|error| format!("fingerprint actual {}: {error}", object.name))?;
if found != wanted {
return Err(format!(
"object `{}` differs: expected {wanted:?}, found {found:?}",
object.name
));
}
}
Ok(())
}
#[derive(Debug, PartialEq, Eq)]
struct CatalogObjectFingerprint {
kind: String,
name: String,
table_name: String,
normalized_sql: Option<String>,
table_columns: Vec<TableColumnFingerprint>,
foreign_keys: Vec<ForeignKeyFingerprint>,
index: Option<IndexFingerprint>,
}
#[derive(Debug, PartialEq, Eq)]
struct TableColumnFingerprint {
cid: i64,
name: String,
declared_type: String,
not_null: bool,
default_value: Option<String>,
primary_key_position: i64,
hidden: i64,
}
#[derive(Debug, PartialEq, Eq)]
struct ForeignKeyFingerprint {
id: i64,
sequence: i64,
target_table: String,
from_column: String,
to_column: Option<String>,
on_update: String,
on_delete: String,
match_clause: String,
}
#[derive(Debug, PartialEq, Eq)]
struct IndexFingerprint {
unique: bool,
origin: String,
partial: bool,
columns: Vec<IndexColumnFingerprint>,
}
#[derive(Debug, PartialEq, Eq)]
struct IndexColumnFingerprint {
sequence: i64,
column_id: i64,
name: Option<String>,
descending: bool,
collation: Option<String>,
key: bool,
}
fn catalog_names(
conn: &Connection,
objects: &[SchemaObject],
) -> Result<Vec<(String, String)>, rusqlite::Error> {
let mut found = Vec::new();
let mut statement = conn.prepare(
"SELECT type, name FROM main.sqlite_schema
WHERE name = ?1 AND name NOT LIKE 'sqlite_%'
ORDER BY type, name",
)?;
for object in objects {
let rows = statement.query_map([object.name], |row| Ok((row.get(0)?, row.get(1)?)))?;
found.extend(rows.collect::<Result<Vec<_>, _>>()?);
}
found.sort();
found.dedup();
Ok(found)
}
fn catalog_fingerprint(
conn: &Connection,
object: &SchemaObject,
) -> Result<CatalogObjectFingerprint, rusqlite::Error> {
let (kind, name, table_name, sql): (String, String, String, Option<String>) = conn.query_row(
"SELECT type, name, tbl_name, sql FROM main.sqlite_schema WHERE name = ?1",
[object.name],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)?;
let (table_columns, foreign_keys) = if kind == "table" || kind == "view" {
(
table_columns(conn, object.name)?,
foreign_keys(conn, object.name)?,
)
} else {
(Vec::new(), Vec::new())
};
let index = if kind == "index" {
Some(index_fingerprint(conn, &table_name, object.name)?)
} else {
None
};
Ok(CatalogObjectFingerprint {
kind,
name,
table_name,
normalized_sql: sql.map(|sql| normalize_schema_sql(&sql)),
table_columns,
foreign_keys,
index,
})
}
fn table_columns(
conn: &Connection,
table: &str,
) -> Result<Vec<TableColumnFingerprint>, rusqlite::Error> {
let mut statement = conn.prepare(
"SELECT cid, name, type, \"notnull\", dflt_value, pk, hidden
FROM pragma_table_xinfo(?1, 'main')
ORDER BY cid",
)?;
let rows = statement.query_map([table], |row| {
Ok(TableColumnFingerprint {
cid: row.get(0)?,
name: row.get(1)?,
declared_type: row.get(2)?,
not_null: row.get(3)?,
default_value: row.get(4)?,
primary_key_position: row.get(5)?,
hidden: row.get(6)?,
})
})?;
rows.collect()
}
fn foreign_keys(
conn: &Connection,
table: &str,
) -> Result<Vec<ForeignKeyFingerprint>, rusqlite::Error> {
let mut statement = conn.prepare(
"SELECT id, seq, \"table\", \"from\", \"to\", on_update, on_delete, \"match\"
FROM pragma_foreign_key_list(?1, 'main')
ORDER BY id, seq",
)?;
let rows = statement.query_map([table], |row| {
Ok(ForeignKeyFingerprint {
id: row.get(0)?,
sequence: row.get(1)?,
target_table: row.get(2)?,
from_column: row.get(3)?,
to_column: row.get(4)?,
on_update: row.get(5)?,
on_delete: row.get(6)?,
match_clause: row.get(7)?,
})
})?;
rows.collect()
}
fn index_fingerprint(
conn: &Connection,
table: &str,
index: &str,
) -> Result<IndexFingerprint, rusqlite::Error> {
let (unique, origin, partial): (bool, String, bool) = conn.query_row(
"SELECT \"unique\", origin, partial
FROM pragma_index_list(?1, 'main')
WHERE name = ?2",
rusqlite::params![table, index],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
let mut statement = conn.prepare(
"SELECT seqno, cid, name, desc, coll, key
FROM pragma_index_xinfo(?1, 'main')
ORDER BY seqno",
)?;
let columns = statement
.query_map([index], |row| {
Ok(IndexColumnFingerprint {
sequence: row.get(0)?,
column_id: row.get(1)?,
name: row.get(2)?,
descending: row.get(3)?,
collation: row.get(4)?,
key: row.get(5)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(IndexFingerprint {
unique,
origin,
partial,
columns,
})
}
fn normalize_schema_sql(sql: &str) -> String {
sql.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.replace(" IF NOT EXISTS ", " ")
}
fn unsupported_predecessor(domain: &SchemaDomain, found: i64) -> SqliteStoreError {
SqliteStoreError::UnsupportedSchemaPredecessor {
domain: domain.name.to_string(),
found,
supported: domain.supported_version(),
allowed: domain.allowed_existing_versions.to_vec(),
}
}
fn find_owned_objects(
conn: &Connection,
domain: &SchemaDomain,
) -> Result<Vec<String>, SqliteStoreError> {
let mut found = Vec::new();
let mut stmt = conn.prepare(
"SELECT type, name FROM main.sqlite_schema
WHERE name = ?1 AND name NOT LIKE 'sqlite_%'
ORDER BY type, name",
)?;
for expected in domain.owned_objects.iter().chain(domain.retired_objects) {
let rows = stmt.query_map([expected.name], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?;
for row in rows {
let (actual_kind, name) = row?;
found.push(format!(
"{actual_kind}:{name} (expected {})",
expected.kind.sqlite_name()
));
}
}
found.sort();
found.dedup();
Ok(found)
}
fn ledger_table_exists(conn: &Connection) -> Result<bool, SqliteStoreError> {
let exists = conn
.query_row(
"SELECT 1 FROM main.sqlite_master WHERE type = 'table' AND name = 'meerkat_schema'",
[],
|_| Ok(()),
)
.optional()?
.is_some();
Ok(exists)
}
fn malformed(detail: String) -> SqliteStoreError {
SqliteStoreError::LedgerMalformed { detail }
}
fn validate_ledger_shape(conn: &Connection) -> Result<(), SqliteStoreError> {
let mut stmt = conn.prepare(
"SELECT name, type, \"notnull\", pk FROM pragma_table_info('meerkat_schema', 'main')",
)?;
let mut rows = stmt.query([])?;
let mut domain_ok = false;
let mut version_ok = false;
while let Some(row) = rows.next()? {
let name: String = row.get(0)?;
let decl_type: String = row.get(1)?;
let notnull: bool = row.get(2)?;
let pk: i64 = row.get(3)?;
match name.as_str() {
"domain" => {
if !decl_type.eq_ignore_ascii_case("TEXT") || pk != 1 {
return Err(malformed(format!(
"column `domain` must be `TEXT PRIMARY KEY`, found type `{decl_type}` \
with pk position {pk}"
)));
}
domain_ok = true;
}
"version" => {
if !decl_type.eq_ignore_ascii_case("INTEGER") || !notnull || pk != 0 {
return Err(malformed(format!(
"column `version` must be non-key `INTEGER NOT NULL`, found type \
`{decl_type}` notnull={notnull} pk position {pk}"
)));
}
version_ok = true;
}
other => {
if pk != 0 {
return Err(malformed(format!(
"unexpected primary-key column `{other}`"
)));
}
}
}
}
if !domain_ok || !version_ok {
return Err(malformed(
"table lacks the pinned `domain`/`version` columns".to_string(),
));
}
Ok(())
}
fn read_version(conn: &Connection, domain: &str) -> Result<Option<i64>, SqliteStoreError> {
let mut stmt = conn.prepare("SELECT version FROM main.meerkat_schema WHERE domain = ?1")?;
let mut rows = stmt.query([domain])?;
let Some(row) = rows.next()? else {
return Ok(None);
};
let version: i64 = row.get(0)?;
if rows.next()?.is_some() {
return Err(malformed(format!(
"multiple ledger rows for domain `{domain}`"
)));
}
if version <= 0 {
return Err(malformed(format!(
"domain `{domain}` records non-positive version {version}"
)));
}
Ok(Some(version))
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
use super::*;
use crate::profile::{ConnectionProfile, open};
fn create_t1(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE IF NOT EXISTS t1 (x INTEGER)")
}
fn add_column_guarded(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
let has_column = tx
.prepare("PRAGMA table_info(t1)")?
.query_map([], |row| row.get::<_, String>(1))?
.collect::<Result<Vec<_>, _>>()?
.iter()
.any(|name| name == "y");
if !has_column {
tx.execute_batch("ALTER TABLE t1 ADD COLUMN y TEXT")?;
}
Ok(())
}
fn initialize_v2(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
create_t1(tx)?;
add_column_guarded(tx)
}
fn initialize_v2_alt(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE t1 (x INTEGER, z BLOB)")
}
const RELEASED_V1_OBJECTS: &[SchemaObject] = &[SchemaObject {
kind: SchemaObjectKind::Table,
name: "t1",
}];
fn verify_v1(conn: &Connection) -> Result<(), String> {
verify_released_schema_fingerprint(conn, &DOMAIN_V2, RELEASED_V1_OBJECTS, create_t1)
}
const DOMAIN_V1: SchemaDomain = SchemaDomain {
name: "test-domain",
migrations: &[Migration {
version: 1,
name: "base",
apply: create_t1,
}],
initialize_current: create_t1,
allowed_existing_versions: &[1],
released_predecessors: &[],
owned_objects: &[SchemaObject {
kind: SchemaObjectKind::Table,
name: "t1",
}],
retired_objects: &[],
};
const DOMAIN_V2: SchemaDomain = SchemaDomain {
name: "test-domain",
migrations: &[
Migration {
version: 1,
name: "base",
apply: create_t1,
},
Migration {
version: 2,
name: "add-y",
apply: add_column_guarded,
},
],
initialize_current: initialize_v2,
allowed_existing_versions: &[1, 2],
released_predecessors: &[SchemaPredecessor {
version: 1,
verify: verify_v1,
}],
owned_objects: &[SchemaObject {
kind: SchemaObjectKind::Table,
name: "t1",
}],
retired_objects: &[],
};
const DOMAIN_V2_ALT_INITIALIZER: SchemaDomain = SchemaDomain {
name: "test-domain",
migrations: &[
Migration {
version: 1,
name: "base",
apply: create_t1,
},
Migration {
version: 2,
name: "alt-current",
apply: add_column_guarded,
},
],
initialize_current: initialize_v2_alt,
allowed_existing_versions: &[2],
released_predecessors: &[],
owned_objects: &[SchemaObject {
kind: SchemaObjectKind::Table,
name: "t1",
}],
retired_objects: &[],
};
fn temp_conn(dir: &tempfile::TempDir) -> Connection {
open(&dir.path().join("db.sqlite3"), ConnectionProfile::PRIMARY).expect("open")
}
#[test]
fn fresh_file_initializes_current_and_stamps() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("apply");
assert_eq!(
report,
LedgerReport {
from_version: 0,
to_version: 2
}
);
assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(2));
conn.execute("INSERT INTO t1 (x, y) VALUES (1, 'a')", [])
.expect("schema converged");
}
#[test]
fn second_open_is_current_noop() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("first");
let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("second");
assert!(!report.migrated());
}
#[test]
fn current_oracle_cache_binds_initializer_and_manifest_not_only_name_version() {
let first_dir = tempfile::tempdir().expect("first tempdir");
let mut first = temp_conn(&first_dir);
apply_domain_migrations(&mut first, &DOMAIN_V2).expect("first current");
let second_dir = tempfile::tempdir().expect("second tempdir");
let mut second = temp_conn(&second_dir);
apply_domain_migrations(&mut second, &DOMAIN_V2_ALT_INITIALIZER).expect("alt current");
let columns: Vec<String> = second
.prepare("PRAGMA table_info(t1)")
.expect("prepare")
.query_map([], |row| row.get(1))
.expect("query")
.collect::<Result<_, _>>()
.expect("columns");
assert_eq!(columns, vec!["x", "z"]);
}
#[test]
fn current_row_with_partial_catalog_is_refused_before_noop() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("current");
conn.execute_batch("ALTER TABLE t1 ADD COLUMN candidate_partial TEXT")
.expect("partial candidate mutation");
let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse current shape");
assert!(matches!(
err,
SqliteStoreError::SchemaFingerprintMismatch { version: 2, .. }
));
assert_eq!(
domain_version(&conn, DOMAIN_V2.name).expect("ledger"),
Some(2)
);
}
#[test]
fn upgrade_applies_only_pending() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("v1");
let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("v2");
assert_eq!(
report,
LedgerReport {
from_version: 1,
to_version: 2
}
);
}
#[test]
fn allowed_version_with_wrong_catalog_is_refused_without_migration() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("v1");
conn.execute_batch("ALTER TABLE t1 ADD COLUMN candidate_only TEXT")
.expect("candidate shape");
let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse fingerprint");
assert!(matches!(
err,
SqliteStoreError::SchemaFingerprintMismatch { version: 1, .. }
));
assert_eq!(
domain_version(&conn, DOMAIN_V1.name).expect("ledger"),
Some(1),
"fingerprint refusal advanced the ledger"
);
let columns: Vec<String> = conn
.prepare("PRAGMA table_info(t1)")
.expect("prepare")
.query_map([], |row| row.get(1))
.expect("query")
.collect::<Result<_, _>>()
.expect("columns");
assert_eq!(columns, vec!["x", "candidate_only"]);
}
#[test]
fn pre_floor_and_gap_versions_are_refused_without_mutation() {
fn no_op(_tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
Ok(())
}
fn verify_v2(_conn: &Connection) -> Result<(), String> {
Ok(())
}
const DOMAIN_V3_FLOOR_2: SchemaDomain = SchemaDomain {
name: "floor-domain",
migrations: &[
Migration {
version: 1,
name: "base",
apply: no_op,
},
Migration {
version: 2,
name: "released-floor",
apply: no_op,
},
Migration {
version: 3,
name: "current",
apply: no_op,
},
],
initialize_current: no_op,
allowed_existing_versions: &[2, 3],
released_predecessors: &[SchemaPredecessor {
version: 2,
verify: verify_v2,
}],
owned_objects: &[],
retired_objects: &[],
};
const DOMAIN_V4_GAP_3: SchemaDomain = SchemaDomain {
name: "gap-domain",
migrations: &[
Migration {
version: 1,
name: "old",
apply: no_op,
},
Migration {
version: 2,
name: "released-floor",
apply: no_op,
},
Migration {
version: 3,
name: "unreleased-candidate",
apply: no_op,
},
Migration {
version: 4,
name: "current",
apply: no_op,
},
],
initialize_current: no_op,
allowed_existing_versions: &[2, 4],
released_predecessors: &[SchemaPredecessor {
version: 2,
verify: verify_v2,
}],
owned_objects: &[],
retired_objects: &[],
};
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
conn.execute_batch(CREATE_LEDGER_SQL).expect("ledger");
conn.execute(
"INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, 1)",
[DOMAIN_V3_FLOOR_2.name],
)
.expect("pre-floor row");
let err =
apply_domain_migrations(&mut conn, &DOMAIN_V3_FLOOR_2).expect_err("refuse pre-floor");
assert!(matches!(
err,
SqliteStoreError::UnsupportedSchemaPredecessor { found: 1, .. }
));
assert_eq!(
domain_version(&conn, DOMAIN_V3_FLOOR_2.name).expect("ledger"),
Some(1)
);
conn.execute(
"INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, 3)",
[DOMAIN_V4_GAP_3.name],
)
.expect("gap row");
let err = apply_domain_migrations(&mut conn, &DOMAIN_V4_GAP_3).expect_err("refuse gap");
assert!(matches!(
err,
SqliteStoreError::UnsupportedSchemaPredecessor { found: 3, .. }
));
assert_eq!(
domain_version(&conn, DOMAIN_V4_GAP_3.name).expect("ledger"),
Some(3)
);
}
#[test]
fn unledgered_owned_objects_are_refused_without_mutation() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
conn.execute_batch("CREATE TABLE t1 (x INTEGER)")
.expect("unknown unledgered ddl");
let err = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect_err("refuse");
assert!(matches!(
err,
SqliteStoreError::UnledgeredDomainObjects { .. }
));
assert!(
!ledger_table_exists(&conn).expect("ledger presence"),
"eligibility refusal must not create the ledger"
);
let columns: Vec<String> = conn
.prepare("PRAGMA table_info(t1)")
.expect("prepare")
.query_map([], |row| row.get(1))
.expect("query")
.collect::<Result<_, _>>()
.expect("columns");
assert_eq!(columns, vec!["x"], "refusal mutated unknown schema");
}
#[test]
fn fresh_domain_ignores_foreign_cotenant_objects() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
conn.execute_batch("CREATE TABLE foreign_table (value TEXT)")
.expect("foreign ddl");
let report = apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("fresh domain");
assert_eq!(
report,
LedgerReport {
from_version: 0,
to_version: 2
}
);
conn.execute("INSERT INTO foreign_table VALUES ('kept')", [])
.expect("foreign object survives");
}
#[test]
fn future_version_is_refused_before_any_mutation() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("stamp v2");
let err = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect_err("refuse");
match err {
SqliteStoreError::SchemaFromTheFuture {
domain,
found,
supported,
} => {
assert_eq!(domain, "test-domain");
assert_eq!(found, 2);
assert_eq!(supported, 1);
}
other => panic!("wrong error: {other}"),
}
assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(2));
}
#[test]
fn foreign_domain_rows_are_untouched() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("mine");
conn.execute(
"INSERT INTO meerkat_schema (domain, version) VALUES ('foreign-domain', 7)",
[],
)
.expect("foreign row");
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("noop");
let foreign: i64 = conn
.query_row(
"SELECT version FROM meerkat_schema WHERE domain = 'foreign-domain'",
[],
|r| r.get(0),
)
.expect("foreign row survives");
assert_eq!(foreign, 7);
}
#[test]
fn invalid_migration_list_is_refused_without_touching_the_file() {
const BAD: SchemaDomain = SchemaDomain {
name: "bad-domain",
migrations: &[Migration {
version: 3,
name: "gap",
apply: create_t1,
}],
initialize_current: create_t1,
allowed_existing_versions: &[3],
released_predecessors: &[],
owned_objects: &[],
retired_objects: &[],
};
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
let err = apply_domain_migrations(&mut conn, &BAD).expect_err("refuse");
assert!(matches!(err, SqliteStoreError::InvalidMigrationList { .. }));
assert!(!ledger_table_exists(&conn).expect("check"));
}
#[test]
fn failed_migration_rolls_back_atomically() {
fn fail(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE half_done (x INTEGER)")?;
tx.execute_batch("THIS IS NOT SQL")
}
fn initialize_failing(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
create_t1(tx)?;
fail(tx)
}
const FAILING: SchemaDomain = SchemaDomain {
name: "failing-domain",
migrations: &[
Migration {
version: 1,
name: "base",
apply: create_t1,
},
Migration {
version: 2,
name: "explodes",
apply: fail,
},
],
initialize_current: initialize_failing,
allowed_existing_versions: &[1, 2],
released_predecessors: &[SchemaPredecessor {
version: 1,
verify: verify_v1,
}],
owned_objects: &[
SchemaObject {
kind: SchemaObjectKind::Table,
name: "t1",
},
SchemaObject {
kind: SchemaObjectKind::Table,
name: "half_done",
},
],
retired_objects: &[],
};
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
let err = apply_domain_migrations(&mut conn, &FAILING).expect_err("must fail");
assert!(matches!(
err,
SqliteStoreError::MigrationFailed { version: 2, .. }
));
assert_eq!(domain_version(&conn, "failing-domain").expect("read"), None);
let tables: Vec<String> = conn
.prepare(
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('t1','half_done')",
)
.expect("prepare")
.query_map([], |r| r.get(0))
.expect("query")
.collect::<Result<_, _>>()
.expect("rows");
assert!(tables.is_empty(), "rollback left tables behind: {tables:?}");
}
#[test]
fn malformed_ledger_shape_is_refused_not_healed() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
conn.execute_batch("CREATE TABLE meerkat_schema (x INTEGER)")
.expect("foreign ddl");
let err = domain_version(&conn, "test-domain").expect_err("refuse read");
assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
let err = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect_err("refuse migrate");
assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM meerkat_schema", [], |r| r.get(0))
.expect("foreign table survives");
assert_eq!(count, 0);
}
#[test]
fn non_positive_versions_are_refused_not_healed() {
for bad_version in [0i64, -3] {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
conn.execute_batch(
"CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL)",
)
.expect("ledger ddl");
conn.execute(
"INSERT INTO meerkat_schema (domain, version) VALUES ('test-domain', ?1)",
[bad_version],
)
.expect("seed bad version");
let err = domain_version(&conn, "test-domain").expect_err("refuse read");
assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
let err = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect_err("refuse migrate");
assert!(matches!(err, SqliteStoreError::LedgerMalformed { .. }));
let stored: i64 = conn
.query_row(
"SELECT version FROM meerkat_schema WHERE domain = 'test-domain'",
[],
|r| r.get(0),
)
.expect("row survives");
assert_eq!(stored, bad_version);
}
}
#[test]
fn duplicate_domain_rows_are_refused() {
let dir = tempfile::tempdir().expect("tempdir");
let conn = temp_conn(&dir);
conn.execute_batch(
"CREATE TABLE meerkat_schema (domain TEXT, version INTEGER NOT NULL);
INSERT INTO meerkat_schema VALUES ('dup-domain', 1);
INSERT INTO meerkat_schema VALUES ('dup-domain', 2);",
)
.expect("seed duplicates");
let err = read_version(&conn, "dup-domain").expect_err("refuse duplicates");
match err {
SqliteStoreError::LedgerMalformed { detail } => {
assert!(detail.contains("multiple ledger rows"), "{detail}");
}
other => panic!("wrong error: {other}"),
}
}
#[test]
fn temp_shadowing_cannot_hijack_the_ledger() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("stamp v1");
conn.execute_batch(
"CREATE TEMP TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
INSERT INTO temp.meerkat_schema VALUES ('test-domain', 999);",
)
.expect("temp shadow");
assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(1));
let report = apply_domain_migrations(&mut conn, &DOMAIN_V1).expect("noop against main");
assert!(!report.migrated());
}
#[test]
fn migration_that_ends_the_transaction_is_refused_unstamped() {
fn no_op(_tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
Ok(())
}
fn verify_empty_predecessor(_conn: &Connection) -> Result<(), String> {
Ok(())
}
fn commits_underneath(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE escaped_commit (x INTEGER); COMMIT")
}
fn rolls_back_underneath(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("ROLLBACK")
}
fn commits_then_begins(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE escaped_commit_begin (x INTEGER); COMMIT; BEGIN")
}
fn rolls_back_then_begins(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("ROLLBACK; BEGIN")
}
const COMMITS: SchemaDomain = SchemaDomain {
name: "custody-commit",
migrations: &[
Migration {
version: 1,
name: "base",
apply: no_op,
},
Migration {
version: 2,
name: "commits-underneath",
apply: commits_underneath,
},
],
initialize_current: no_op,
allowed_existing_versions: &[1, 2],
released_predecessors: &[SchemaPredecessor {
version: 1,
verify: verify_empty_predecessor,
}],
owned_objects: &[],
retired_objects: &[],
};
const ROLLS_BACK: SchemaDomain = SchemaDomain {
name: "custody-rollback",
migrations: &[
Migration {
version: 1,
name: "base",
apply: no_op,
},
Migration {
version: 2,
name: "rolls-back-underneath",
apply: rolls_back_underneath,
},
],
initialize_current: no_op,
allowed_existing_versions: &[1, 2],
released_predecessors: &[SchemaPredecessor {
version: 1,
verify: verify_empty_predecessor,
}],
owned_objects: &[],
retired_objects: &[],
};
const COMMITS_THEN_BEGINS: SchemaDomain = SchemaDomain {
name: "custody-commit-begin",
migrations: &[
Migration {
version: 1,
name: "base",
apply: no_op,
},
Migration {
version: 2,
name: "commits-then-begins",
apply: commits_then_begins,
},
],
initialize_current: no_op,
allowed_existing_versions: &[1, 2],
released_predecessors: &[SchemaPredecessor {
version: 1,
verify: verify_empty_predecessor,
}],
owned_objects: &[],
retired_objects: &[],
};
const ROLLS_BACK_THEN_BEGINS: SchemaDomain = SchemaDomain {
name: "custody-rollback-begin",
migrations: &[
Migration {
version: 1,
name: "base",
apply: no_op,
},
Migration {
version: 2,
name: "rolls-back-then-begins",
apply: rolls_back_then_begins,
},
],
initialize_current: no_op,
allowed_existing_versions: &[1, 2],
released_predecessors: &[SchemaPredecessor {
version: 1,
verify: verify_empty_predecessor,
}],
owned_objects: &[],
retired_objects: &[],
};
for (domain, expected_name) in [
(&COMMITS, "commits-underneath"),
(&ROLLS_BACK, "rolls-back-underneath"),
(&COMMITS_THEN_BEGINS, "commits-then-begins"),
(&ROLLS_BACK_THEN_BEGINS, "rolls-back-then-begins"),
] {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
conn.execute_batch(CREATE_LEDGER_SQL).expect("ledger");
conn.execute(
"INSERT INTO main.meerkat_schema (domain, version) VALUES (?1, 1)",
[domain.name],
)
.expect("released predecessor");
let err = apply_domain_migrations(&mut conn, domain).expect_err("custody violation");
match err {
SqliteStoreError::MigrationBrokeTransaction {
domain: err_domain,
version,
name,
} => {
assert_eq!(err_domain, domain.name);
assert_eq!(version, 2);
assert_eq!(name, expected_name);
}
other => panic!("wrong error: {other}"),
}
assert_eq!(domain_version(&conn, domain.name).expect("read"), Some(1));
}
}
#[test]
fn initializer_that_ends_the_transaction_is_refused_unstamped() {
fn commits_underneath(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch("CREATE TABLE escaped_initializer (x INTEGER); COMMIT")
}
const COMMITS: SchemaDomain = SchemaDomain {
name: "initializer-custody-commit",
migrations: &[Migration {
version: 1,
name: "base",
apply: commits_underneath,
}],
initialize_current: commits_underneath,
allowed_existing_versions: &[1],
released_predecessors: &[],
owned_objects: &[SchemaObject {
kind: SchemaObjectKind::Table,
name: "escaped_initializer",
}],
retired_objects: &[],
};
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
let err = apply_domain_migrations(&mut conn, &COMMITS).expect_err("custody violation");
match err {
SqliteStoreError::MigrationBrokeTransaction {
domain,
version,
name,
} => {
assert_eq!(domain, COMMITS.name);
assert_eq!(version, 1);
assert_eq!(name, "initialize-current");
}
other => panic!("wrong error: {other}"),
}
assert_eq!(domain_version(&conn, COMMITS.name).expect("read"), None);
}
#[test]
fn initializer_using_its_own_savepoints_keeps_custody() {
fn nests_savepoints(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch(
"SAVEPOINT body_sp;
CREATE TABLE sp_t (x INTEGER);
RELEASE SAVEPOINT body_sp",
)
}
const NESTED: SchemaDomain = SchemaDomain {
name: "custody-nested-savepoint",
migrations: &[Migration {
version: 1,
name: "nests-savepoints",
apply: nests_savepoints,
}],
initialize_current: nests_savepoints,
allowed_existing_versions: &[1],
released_predecessors: &[],
owned_objects: &[SchemaObject {
kind: SchemaObjectKind::Table,
name: "sp_t",
}],
retired_objects: &[],
};
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
let report = apply_domain_migrations(&mut conn, &NESTED).expect("apply");
assert_eq!(report.to_version, 1);
assert_eq!(domain_version(&conn, NESTED.name).expect("read"), Some(1));
}
#[test]
fn schema_preflight_passes_fresh_and_current_refuses_future() {
let dir = tempfile::tempdir().expect("tempdir");
let mut conn = temp_conn(&dir);
preflight_schema_eligibility(&conn, &DOMAIN_V1).expect("no ledger yet");
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("stamp v2");
preflight_schema_eligibility(&conn, &DOMAIN_V2).expect("current");
let err =
preflight_schema_eligibility(&conn, &DOMAIN_V1).expect_err("future for old binary");
assert!(matches!(
err,
SqliteStoreError::SchemaFromTheFuture {
found: 2,
supported: 1,
..
}
));
}
#[test]
fn concurrent_opens_race_safely() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("db.sqlite3");
let mut handles = Vec::new();
for _ in 0..8 {
let path = path.clone();
handles.push(std::thread::spawn(move || {
let mut conn = open(&path, ConnectionProfile::PRIMARY).expect("open");
apply_domain_migrations(&mut conn, &DOMAIN_V2).expect("apply")
}));
}
let mut migrated = 0;
for handle in handles {
let report = handle.join().expect("thread");
assert_eq!(report.to_version, 2);
if report.migrated() {
migrated += 1;
}
}
assert!(migrated >= 1, "someone must have migrated");
let conn = open(&path, ConnectionProfile::ReadOnly).expect("reopen");
assert_eq!(domain_version(&conn, "test-domain").expect("read"), Some(2));
}
}