use crate::recover::{
allowed, is_unreadable, quarantine_database_journaled, unfinished_quarantine, RecoveryPolicy,
QUARANTINE_JOURNAL_SUFFIX,
};
use crate::{DbError, DbOptions, Result};
use rusqlite::{Connection, ErrorCode, OptionalExtension};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
pub enum Step {
Sql(&'static str),
Code(fn(&Connection) -> rusqlite::Result<()>),
}
pub struct Migration {
pub version: i64,
pub name: &'static str,
pub up: Step,
pub down: Option<&'static str>,
}
impl Migration {
pub const fn sql(version: i64, name: &'static str, up: &'static str) -> Self {
Self {
version,
name,
up: Step::Sql(up),
down: None,
}
}
pub const fn with_down(mut self, down: &'static str) -> Self {
self.down = Some(down);
self
}
pub const fn code(
version: i64,
name: &'static str,
up: fn(&Connection) -> rusqlite::Result<()>,
) -> Self {
Self {
version,
name,
up: Step::Code(up),
down: None,
}
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub enum VersionStorage {
#[default]
UserVersion,
Table { table: String, key: String },
VersionTable {
table: String,
version_column: String,
applied_at_column: String,
},
}
impl VersionStorage {
pub fn version_table(table: impl Into<String>) -> Self {
Self::VersionTable {
table: table.into(),
version_column: "version".into(),
applied_at_column: "applied_at".into(),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LegacyVersionTable {
pub table: String,
pub version_column: String,
}
pub type VerifyBackup = fn(&Path) -> std::result::Result<(), String>;
#[derive(Clone)]
pub struct MigrateOptions {
pub version_storage: VersionStorage,
pub backup: bool,
pub backup_options: Option<DbOptions>,
pub forward_compatible: Option<fn(&Connection) -> bool>,
pub optimize_after: bool,
pub legacy_versions: Option<LegacyVersionTable>,
pub backup_name: Option<fn(current: i64, target: i64) -> String>,
pub verify_backup: Option<VerifyBackup>,
pub backup_retention: Option<usize>,
}
impl Default for MigrateOptions {
fn default() -> Self {
Self {
version_storage: VersionStorage::default(),
backup: true,
backup_options: None,
forward_compatible: None,
optimize_after: true,
legacy_versions: None,
backup_name: None,
verify_backup: None,
backup_retention: None,
}
}
}
impl MigrateOptions {
pub fn import_legacy_versions(mut self, legacy: LegacyVersionTable) -> Self {
self.legacy_versions = Some(legacy);
self
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct MigrationReport {
pub from: i64,
pub to: i64,
pub applied: Vec<i64>,
pub backup: Option<PathBuf>,
pub forward_compatible: bool,
pub interrupted_previous: Option<PendingMarker>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PendingMarker {
pub from: i64,
pub to: i64,
pub pid: u32,
}
pub fn user_version(conn: &Connection) -> Result<i64> {
Ok(conn.query_row("PRAGMA user_version", [], |row| row.get(0))?)
}
fn set_user_version(conn: &Connection, version: i64) -> rusqlite::Result<()> {
conn.execute_batch(&format!("PRAGMA user_version = {version};"))
}
pub fn migration_version(conn: &Connection, storage: &VersionStorage) -> Result<i64> {
match storage {
VersionStorage::UserVersion => user_version(conn),
VersionStorage::Table { table, key } => key_value_version(conn, table, key),
VersionStorage::VersionTable {
table,
version_column,
..
} => version_row_max(conn, table, version_column),
}
}
fn key_value_version(conn: &Connection, table: &str, key: &str) -> Result<i64> {
check_ident(table)?;
let exists: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 COLLATE NOCASE)",
[table],
|row| row.get(0),
)?;
if !exists {
return Ok(0);
}
let value: Option<String> = conn
.query_row(
&format!("SELECT CAST(\"value\" AS TEXT) FROM \"{table}\" WHERE \"key\"=?1"),
[key],
|row| row.get(0),
)
.optional()?;
match value {
None => Ok(0),
Some(value) => value
.parse::<i64>()
.ok()
.filter(|version| *version >= 0)
.ok_or_else(|| DbError::InvalidMigrations(format!("invalid stored version {value:?}"))),
}
}
fn set_key_value_version(
conn: &Connection,
storage: &VersionStorage,
version: i64,
) -> rusqlite::Result<()> {
match storage {
VersionStorage::UserVersion => set_user_version(conn, version),
VersionStorage::Table { table, key } => {
conn.execute_batch(&format!(
"CREATE TABLE IF NOT EXISTS \"{table}\" (\"key\" TEXT PRIMARY KEY, \"value\" TEXT NOT NULL);"
))?;
conn.execute(
&format!(
"INSERT INTO \"{table}\"(\"key\",\"value\") VALUES(?1,?2) \
ON CONFLICT(\"key\") DO UPDATE SET \"value\"=excluded.\"value\""
),
rusqlite::params![key, version.to_string()],
)?;
Ok(())
}
VersionStorage::VersionTable { .. } => Err(rusqlite::Error::InvalidQuery),
}
}
fn advance_version(
conn: &Connection,
storage: &VersionStorage,
version: i64,
) -> rusqlite::Result<()> {
match storage {
VersionStorage::VersionTable {
table,
version_column,
applied_at_column,
} => {
conn.execute_batch(&format!(
"CREATE TABLE IF NOT EXISTS \"{table}\" (\"{version_column}\" INTEGER PRIMARY KEY, \"{applied_at_column}\" INTEGER NOT NULL);"
))?;
conn.execute(
&format!("DELETE FROM \"{table}\" WHERE \"{version_column}\" = ?1"),
[version],
)?;
conn.execute(
&format!(
"INSERT INTO \"{table}\" (\"{version_column}\", \"{applied_at_column}\") VALUES (?1, ?2)"
),
rusqlite::params![version, now_secs()],
)?;
Ok(())
}
_ => set_key_value_version(conn, storage, version),
}
}
fn retreat_version(
conn: &Connection,
storage: &VersionStorage,
version: i64,
) -> rusqlite::Result<()> {
match storage {
VersionStorage::VersionTable {
table,
version_column,
..
} => {
conn.execute(
&format!("DELETE FROM \"{table}\" WHERE \"{version_column}\" = ?1"),
[version],
)?;
Ok(())
}
_ => set_key_value_version(conn, storage, version - 1),
}
}
fn adopt_legacy_version(conn: &Connection, storage: &VersionStorage, through: i64) -> Result<()> {
conn.execute_batch("BEGIN IMMEDIATE;")?;
let seeded = match storage {
VersionStorage::VersionTable { .. } => {
(1..=through).try_for_each(|version| advance_version(conn, storage, version))
}
_ => advance_version(conn, storage, through),
};
match seeded {
Ok(()) => {
conn.execute_batch("COMMIT;")?;
Ok(())
}
Err(error) => {
let _ = conn.execute_batch("ROLLBACK;");
Err(DbError::Sqlite(error))
}
}
}
fn version_row_max(conn: &Connection, table: &str, version_column: &str) -> Result<i64> {
check_ident(table)?;
check_ident(version_column)?;
let exists: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1 COLLATE NOCASE)",
[table],
|row| row.get(0),
)?;
if !exists {
return Ok(0);
}
let highest: Option<i64> = conn.query_row(
&format!("SELECT MAX(CAST(\"{version_column}\" AS INTEGER)) FROM \"{table}\""),
[],
|row| row.get(0),
)?;
let version = highest.unwrap_or(0);
if version < 0 {
return Err(DbError::InvalidMigrations(format!(
"invalid stored version {version}"
)));
}
Ok(version)
}
fn check_storage(storage: &VersionStorage) -> Result<()> {
match storage {
VersionStorage::UserVersion => Ok(()),
VersionStorage::Table { table, .. } => check_ident(table),
VersionStorage::VersionTable {
table,
version_column,
applied_at_column,
} => {
check_ident(table)?;
check_ident(version_column)?;
check_ident(applied_at_column)
}
}
}
fn backup_path(db: &Path, options: &MigrateOptions, current: i64, target: i64) -> Result<PathBuf> {
let name = db
.file_name()
.ok_or(DbError::InvalidDatabasePath)?
.to_string_lossy()
.into_owned();
let file = match options.backup_name {
Some(namer) => namer(current, target),
None => format!("{name}.pre-v{target}.bak"),
};
if file.is_empty() || file.contains(['/', '\\', '\0']) || file == "." || file == ".." {
return Err(DbError::Backup(format!(
"backup name {file:?} must be a plain file name other than the database's"
)));
}
let lowered = file.to_lowercase();
let reserved = [
"",
"-wal",
"-shm",
"-journal",
".migrating",
".restore-journal",
".store-lock",
QUARANTINE_JOURNAL_SUFFIX,
]
.iter()
.any(|suffix| lowered == format!("{name}{suffix}").to_lowercase());
if reserved {
return Err(DbError::Backup(format!(
"backup name {file:?} is the database, one of its sidecars or a recovery file; choose another name"
)));
}
Ok(db.with_file_name(file))
}
fn prune_backups(db: &Path, newest: &Path, keep: usize) {
let Some(name) = db.file_name().map(|n| n.to_string_lossy().into_owned()) else {
return;
};
let dir = db
.parent()
.filter(|p| !p.as_os_str().is_empty())
.unwrap_or(Path::new("."));
let prefix = format!("{name}.pre-");
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
let mut older: Vec<(SystemTime, PathBuf)> = entries
.filter_map(|entry| entry.ok().map(|entry| entry.path()))
.filter(|path| path.as_path() != newest)
.filter(|path| {
path.file_name().is_some_and(|n| {
let n = n.to_string_lossy();
n.starts_with(&prefix) && n.ends_with(".bak")
})
})
.filter_map(|path| {
let modified = std::fs::metadata(&path).and_then(|m| m.modified()).ok()?;
Some((modified, path))
})
.collect();
older.sort_by_key(|(modified, _)| std::cmp::Reverse(*modified));
for (_, path) in older.into_iter().skip(keep.saturating_sub(1)) {
let _ = std::fs::remove_file(path);
}
}
fn now_secs() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
pub fn validate(migrations: &[Migration]) -> Result<()> {
for (index, migration) in migrations.iter().enumerate() {
let expected = index as i64 + 1;
if migration.version != expected {
return Err(DbError::InvalidMigrations(format!(
"expected version {expected} at position {index}, found {} ({})",
migration.version, migration.name
)));
}
}
Ok(())
}
fn marker_path(db: &Path) -> PathBuf {
let mut name = db.as_os_str().to_os_string();
name.push(".migrating");
PathBuf::from(name)
}
pub fn pending_marker(db: &Path) -> Option<PendingMarker> {
let text = std::fs::read_to_string(marker_path(db)).ok()?;
let mut marker = PendingMarker {
from: 0,
to: 0,
pid: 0,
};
for line in text.lines() {
match line.split_once('=')? {
("from", v) => marker.from = v.parse().ok()?,
("to", v) => marker.to = v.parse().ok()?,
("pid", v) => marker.pid = v.parse().ok()?,
_ => {}
}
}
Some(marker)
}
pub fn pending_marker_started_at(db: &Path) -> Option<u64> {
pending_marker(db)?;
let text = std::fs::read_to_string(marker_path(db)).ok()?;
Some(
text.lines()
.find_map(|line| line.strip_prefix("started="))
.and_then(|v| v.parse().ok())
.unwrap_or(0),
)
}
pub fn write_pending_marker(db: &Path, from: i64, to: i64) -> Result<()> {
let body = format!(
"from={from}\nto={to}\npid={}\nstarted={}\n",
std::process::id(),
now_secs().max(0)
);
rightkit_fs::atomic_write(&marker_path(db), body.as_bytes())?;
Ok(())
}
pub fn stored_version_is_newer(
conn: &Connection,
storage: &VersionStorage,
migrations: &[Migration],
) -> Result<bool> {
validate(migrations)?;
check_storage(storage)?;
let max = migrations.last().map_or(0, |m| m.version);
Ok(migration_version(conn, storage)? > max)
}
pub fn migrate(
conn: &Connection,
db_path: Option<&Path>,
migrations: &[Migration],
options: &MigrateOptions,
) -> Result<MigrationReport> {
validate(migrations)?;
check_storage(&options.version_storage)?;
let mut current = migration_version(conn, &options.version_storage)?;
let max = migrations.last().map(|m| m.version).unwrap_or(0);
let mut report = MigrationReport {
from: current,
to: current,
..Default::default()
};
report.interrupted_previous = db_path.and_then(pending_marker);
if let Some(legacy) = options.legacy_versions.as_ref().filter(|_| current == 0) {
let legacy_version = version_row_max(conn, &legacy.table, &legacy.version_column)?;
if legacy_version > 0 {
if legacy_version <= max {
adopt_legacy_version(conn, &options.version_storage, legacy_version)?;
}
current = legacy_version;
report.from = current;
report.to = current;
}
}
if current > max {
if options.forward_compatible.is_some_and(|probe| probe(conn)) {
report.forward_compatible = true;
return Ok(report);
}
return Err(DbError::NewerDatabase { current, max });
}
let pending: Vec<&Migration> = migrations.iter().filter(|m| m.version > current).collect();
if pending.is_empty() {
return Ok(report);
}
if current > 0 && options.backup {
if let Some(path) = db_path {
let backup = backup_path(path, options, current, max)?;
let backup_options = options
.backup_options
.clone()
.unwrap_or_else(|| DbOptions::new(path));
crate::online_backup_to(&backup_options, conn, &backup, None)?;
if let Some(verify) = options.verify_backup {
verify(&backup).map_err(|reason| {
DbError::Backup(format!("pre-migration backup failed its check: {reason}"))
})?;
}
if let Some(keep) = options.backup_retention {
prune_backups(path, &backup, keep);
}
report.backup = Some(backup);
}
}
if let Some(path) = db_path {
write_pending_marker(path, current, max)?;
}
for migration in pending {
conn.execute_batch("BEGIN IMMEDIATE;")?;
let applied = match &migration.up {
Step::Sql(sql) => conn.execute_batch(sql),
Step::Code(run) => run(conn),
}
.and_then(|()| advance_version(conn, &options.version_storage, migration.version));
match applied {
Ok(()) => {
conn.execute_batch("COMMIT;")?;
report.applied.push(migration.version);
report.to = migration.version;
}
Err(source) => {
let _ = conn.execute_batch("ROLLBACK;");
return Err(DbError::MigrationFailed {
version: migration.version,
name: migration.name,
source,
});
}
}
}
if let Some(path) = db_path {
let _ = std::fs::remove_file(marker_path(path));
}
if options.optimize_after {
crate::optimize_boundary(conn)?;
}
Ok(report)
}
pub fn rollback_to(conn: &Connection, migrations: &[Migration], target: i64) -> Result<Vec<i64>> {
rollback_to_with_storage(conn, migrations, target, &VersionStorage::UserVersion)
}
pub fn rollback_to_with_storage(
conn: &Connection,
migrations: &[Migration],
target: i64,
storage: &VersionStorage,
) -> Result<Vec<i64>> {
validate(migrations)?;
check_storage(storage)?;
if target < 0 {
return Err(DbError::Rollback(
"target version must be nonnegative".into(),
));
}
let current = migration_version(conn, storage)?;
if current <= target {
return Err(DbError::Rollback(format!(
"already at v{current}, at or below target v{target}"
)));
}
let mut plan = Vec::new();
for version in ((target + 1)..=current).rev() {
let migration = migrations
.iter()
.find(|m| m.version == version)
.ok_or_else(|| DbError::Rollback(format!("v{version} is not a known migration")))?;
let down = migration.down.ok_or_else(|| {
DbError::Rollback(format!(
"v{version} ({}) has no down migration",
migration.name
))
})?;
plan.push((version, down));
}
let mut reverted = Vec::new();
for (version, down) in plan {
conn.execute_batch("BEGIN IMMEDIATE;")?;
let result = conn
.execute_batch(down)
.and_then(|()| retreat_version(conn, storage, version));
match result {
Ok(()) => {
conn.execute_batch("COMMIT;")?;
reverted.push(version);
}
Err(error) => {
let _ = conn.execute_batch("ROLLBACK;");
return Err(DbError::Rollback(format!(
"down v{version} failed and was rolled back: {error}"
)));
}
}
}
Ok(reverted)
}
pub fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
check_ident(table)?;
let mut statement = conn.prepare(&format!("PRAGMA table_xinfo({table})"))?;
let names = statement.query_map([], |row| row.get::<_, String>(1))?;
for name in names {
if name? == column {
return Ok(true);
}
}
Ok(false)
}
pub fn add_column_if_missing(
conn: &Connection,
table: &str,
column: &str,
decl: &str,
) -> Result<()> {
check_ident(table)?;
check_ident(column)?;
if !column_exists(conn, table, column)? {
conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {decl};"))?;
}
Ok(())
}
fn check_ident(ident: &str) -> Result<()> {
if !ident.is_empty()
&& ident
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'_')
{
Ok(())
} else {
Err(DbError::InvalidMigrations(format!(
"unsafe SQL identifier {ident:?}"
)))
}
}
#[derive(Debug)]
pub struct OpenOrDegrade {
pub conn: Connection,
pub report: MigrationReport,
pub degraded: Option<Degraded>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Degraded {
pub reason: String,
pub quarantined: Option<PathBuf>,
pub quarantine_incomplete: bool,
pub in_memory: bool,
}
impl DbOptions {
pub fn open_writer_migrated(
&self,
migrations: &[Migration],
options: &MigrateOptions,
) -> Result<(Connection, MigrationReport)> {
let conn = self.open_writer()?;
let mut options = options.clone();
options.backup_options.get_or_insert_with(|| self.clone());
let report = migrate(&conn, Some(&self.path), migrations, &options)?;
Ok((conn, report))
}
pub fn open_writer_migrated_or_degrade(
&self,
migrations: &[Migration],
options: &MigrateOptions,
policy: RecoveryPolicy,
) -> Result<OpenOrDegrade> {
if let Some(journal) = unfinished_quarantine(&self.path) {
let reason = format!(
"an earlier quarantine of {} stopped part way and was not undone; see {}",
self.path.display(),
journal.display()
);
let refuse = DbError::Fs(rightkit_fs::FsError::io(
&journal,
std::io::Error::other(reason.clone()),
));
return self.degrade_in_memory(migrations, options, policy, reason, None, true, refuse);
}
let error = match self.open_writer_migrated(migrations, options) {
Ok((conn, report)) => {
return Ok(OpenOrDegrade {
conn,
report,
degraded: None,
})
}
Err(error) if is_recoverable_damage(&error) => error,
Err(error) => return Err(error),
};
let reason = error.to_string();
if !allowed(self, policy) {
return self.degrade_in_memory(migrations, options, policy, reason, None, false, error);
}
let quarantined = match quarantine_database_journaled(
&self.path,
policy.quarantine_stamp,
policy.sidecars,
) {
Ok(moved) => moved,
Err(failure) => {
let reason = format!("{reason}; quarantine failed: {}", failure.error);
let incomplete = failure.stranded_at.is_some();
return self.degrade_in_memory(
migrations,
options,
policy,
reason,
failure.stranded_at,
incomplete,
failure.error,
);
}
};
match self.open_writer_migrated(migrations, options) {
Ok((conn, report)) => Ok(OpenOrDegrade {
conn,
report,
degraded: Some(Degraded {
reason,
quarantined: Some(quarantined),
quarantine_incomplete: false,
in_memory: false,
}),
}),
Err(second) => {
let reason = format!("{reason}; reopen after quarantine failed: {second}");
self.degrade_in_memory(
migrations,
options,
policy,
reason,
Some(quarantined),
false,
second,
)
}
}
}
#[allow(clippy::too_many_arguments)]
fn degrade_in_memory(
&self,
migrations: &[Migration],
options: &MigrateOptions,
policy: RecoveryPolicy,
reason: String,
quarantined: Option<PathBuf>,
quarantine_incomplete: bool,
refuse: DbError,
) -> Result<OpenOrDegrade> {
if !policy.allow_in_memory_fallback {
return Err(refuse);
}
let conn = self.open_in_memory()?;
let report = migrate(&conn, None, migrations, options)?;
Ok(OpenOrDegrade {
conn,
report,
degraded: Some(Degraded {
reason,
quarantined,
quarantine_incomplete,
in_memory: true,
}),
})
}
}
fn is_recoverable_damage(error: &DbError) -> bool {
match error {
DbError::MigrationFailed {
source: rusqlite::Error::SqliteFailure(failure, _),
..
} => matches!(
failure.code,
ErrorCode::NotADatabase | ErrorCode::DatabaseCorrupt
),
other => is_unreadable(other),
}
}