use sqlx::SqlitePool;
use sqlx::migrate::Migrator;
use super::DbError;
pub fn migrador() -> Migrator {
sqlx::migrate!()
}
pub async fn migrar(pool: &SqlitePool) -> Result<(), DbError> {
aplicar(pool, &migrador()).await
}
pub async fn aplicar(pool: &SqlitePool, migrator: &Migrator) -> Result<(), DbError> {
let mut conexion = pool
.acquire()
.await
.map_err(|causa| DbError::EsquemaNoAplicado { causa })?;
pragma_de_la_conexion(&mut conexion, PRAGMA_SIN_CLAVES_FORANEAS).await?;
let resultado = migrator
.run_direct(&mut *conexion)
.await
.map_err(|causa| DbError::Migracion { causa });
pragma_de_la_conexion(&mut conexion, PRAGMA_CON_CLAVES_FORANEAS).await?;
resultado?;
exigir_que_no_queden_huerfanos(&mut conexion).await
}
const PRAGMA_SIN_CLAVES_FORANEAS: &str = "PRAGMA foreign_keys = OFF";
const PRAGMA_CON_CLAVES_FORANEAS: &str = "PRAGMA foreign_keys = ON";
async fn pragma_de_la_conexion(
conexion: &mut sqlx::SqliteConnection,
pragma: &'static str,
) -> Result<(), DbError> {
sqlx::query(pragma)
.execute(conexion)
.await
.map(|_| ())
.map_err(|causa| DbError::EsquemaNoAplicado { causa })
}
async fn exigir_que_no_queden_huerfanos(
conexion: &mut sqlx::SqliteConnection,
) -> Result<(), DbError> {
let tablas = sqlx::query_scalar::<_, String>(
"SELECT DISTINCT \"table\" FROM pragma_foreign_key_check ORDER BY \"table\"",
)
.fetch_all(conexion)
.await
.map_err(|causa| DbError::EsquemaNoAplicado { causa })?;
if tablas.is_empty() {
return Ok(());
}
Err(DbError::HuerfanosTrasMigrar {
tablas: tablas.join(", "),
})
}
#[cfg(test)]
mod tests {
use std::borrow::Cow;
use sqlx::Row;
use sqlx::SqlitePool;
use sqlx::migrate::Migrator;
use crate::db::{abrir, preparar, preparar_con};
use crate::test_support::{DirectorioDePrueba, config_con_dsn, directorio_de_prueba};
async fn base_migrada() -> (DirectorioDePrueba, SqlitePool) {
let directorio = directorio_de_prueba();
let pool = preparar(&config_con_dsn(&directorio.dsn()))
.await
.expect("preparar debe dejar el esquema aplicado");
(directorio, pool)
}
fn migrador_hasta(version: i64) -> Migrator {
let mut migraciones = super::migrador().migrations.into_owned();
migraciones.retain(|migracion| migracion.version <= version);
Migrator {
migrations: Cow::Owned(migraciones),
..super::migrador()
}
}
async fn base_hasta(version: i64) -> (DirectorioDePrueba, SqlitePool) {
let directorio = directorio_de_prueba();
let pool = preparar_con(&config_con_dsn(&directorio.dsn()), &migrador_hasta(version))
.await
.expect("preparar debe dejar el esquema hasta esa versión aplicado");
(directorio, pool)
}
async fn columnas_de(pool: &SqlitePool, tabla: &str) -> Vec<String> {
sqlx::query("SELECT name FROM pragma_table_info(?1)")
.bind(tabla)
.fetch_all(pool)
.await
.expect("la consulta de columnas no puede fallar")
.iter()
.map(|fila| fila.try_get::<String, _>("name").expect("name es texto"))
.collect()
}
#[derive(Debug, PartialEq)]
struct Columna {
nombre: String,
tipo: String,
obligatoria: bool,
por_defecto: Option<String>,
}
fn columna(nombre: &str, tipo: &str, obligatoria: bool, por_defecto: Option<&str>) -> Columna {
Columna {
nombre: nombre.to_owned(),
tipo: tipo.to_owned(),
obligatoria,
por_defecto: por_defecto.map(ToOwned::to_owned),
}
}
async fn columnas_completas_de(pool: &SqlitePool, tabla: &str) -> Vec<Columna> {
sqlx::query("SELECT name, type, \"notnull\", dflt_value FROM pragma_table_info(?1)")
.bind(tabla)
.fetch_all(pool)
.await
.expect("la consulta de columnas no puede fallar")
.iter()
.map(|fila| Columna {
nombre: fila.try_get::<String, _>("name").expect("name es texto"),
tipo: fila.try_get::<String, _>("type").expect("type es texto"),
obligatoria: fila
.try_get::<i64, _>("notnull")
.expect("notnull es un entero")
!= 0,
por_defecto: fila
.try_get::<Option<String>, _>("dflt_value")
.expect("dflt_value es texto o nulo"),
})
.collect()
}
async fn plan_de(pool: &SqlitePool, consulta: &str) -> String {
let filas = sqlx::query(consulta)
.fetch_all(pool)
.await
.unwrap_or_else(|error| panic!("el plan de `{consulta}` no puede fallar: {error}"));
assert!(
!filas.is_empty(),
"`{consulta}` no devolvió ninguna línea de plan"
);
filas
.iter()
.map(|fila| {
fila.try_get::<String, _>("detail")
.expect("el plan trae la columna detail")
})
.collect::<Vec<_>>()
.join("\n")
}
#[tokio::test]
async fn las_cuatro_tablas_tienen_las_columnas_del_srs() {
let (_directorio, pool) = base_migrada().await;
let esperadas: [(&str, Vec<Columna>); 4] = [
(
"repositories",
vec![
columna("id", "TEXT", true, None),
columna("name", "TEXT", true, None),
columna("endpoint", "TEXT", true, None),
columna("bucket", "TEXT", true, None),
columna("region", "TEXT", true, None),
columna("access_key_id", "TEXT", true, None),
columna("secret_access_key", "TEXT", true, None),
columna("passphrase", "TEXT", true, None),
columna("created_at", "DATETIME", true, Some("CURRENT_TIMESTAMP")),
],
),
(
"backup_targets",
vec![
columna("id", "TEXT", true, None),
columna("name", "TEXT", true, None),
columna("target_type", "TEXT", true, None),
columna("container_name", "TEXT", false, None),
columna("db_user", "TEXT", false, None),
columna("db_password", "TEXT", false, None),
columna("db_name", "TEXT", false, None),
columna("host_path", "TEXT", false, None),
columna("db_path", "TEXT", false, None),
columna("created_at", "DATETIME", true, Some("CURRENT_TIMESTAMP")),
],
),
(
"backup_jobs",
vec![
columna("id", "TEXT", true, None),
columna("target_id", "TEXT", true, None),
columna("repository_id", "TEXT", true, None),
columna("cron_expression", "TEXT", true, None),
columna("keep_daily", "INTEGER", true, Some("7")),
columna("keep_weekly", "INTEGER", true, Some("4")),
columna("keep_monthly", "INTEGER", true, Some("12")),
columna("enabled", "BOOLEAN", true, Some("1")),
columna("created_at", "DATETIME", true, Some("CURRENT_TIMESTAMP")),
columna("name", "TEXT", false, None),
],
),
(
"job_executions",
vec![
columna("id", "TEXT", true, None),
columna("job_id", "TEXT", true, None),
columna("status", "TEXT", true, None),
columna("snapshot_id", "TEXT", false, None),
columna("bytes_processed", "INTEGER", true, Some("0")),
columna("bytes_uploaded", "INTEGER", true, Some("0")),
columna("error_message", "TEXT", false, None),
columna("started_at", "DATETIME", true, None),
columna("finished_at", "DATETIME", false, None),
],
),
];
for (tabla, columnas) in esperadas {
let declaradas = columnas_completas_de(&pool, tabla).await;
assert_eq!(
declaradas, columnas,
"las columnas de {tabla} no son las del §3 (nombre, tipo, NOT NULL o \
valor por defecto)"
);
}
pool.close().await;
}
#[tokio::test]
async fn la_0005_anade_el_nombre_sin_reconstruir_backup_jobs() {
let (_directorio, pool) = base_hasta(4).await;
for sentencia in [
"INSERT INTO repositories (id, name, endpoint, bucket, region, access_key_id, \
secret_access_key, passphrase) VALUES ('r-1', 'repo', 'e', 'b', 'r', 'a', 's', 'p')",
"INSERT INTO backup_targets (id, name, target_type, host_path) \
VALUES ('t-1', 'carpeta', 'directory', '/srv/datos')",
"INSERT INTO backup_jobs (id, target_id, repository_id, cron_expression) \
VALUES ('j-1', 't-1', 'r-1', '0 3 * * *')",
"INSERT INTO backup_jobs (id, target_id, repository_id, cron_expression) \
VALUES ('j-2', 't-1', 'r-1', '0 4 * * *')",
] {
sqlx::query(sentencia)
.execute(&pool)
.await
.expect("la siembra de la prueba del nombre no puede fallar");
}
let antes = columnas_de(&pool, "backup_jobs").await;
assert!(
!antes.iter().any(|columna| columna == "name"),
"la columna `name` no puede existir antes de la `0005`: {antes:?}"
);
super::aplicar(&pool, &super::migrador())
.await
.expect("la `0005` tiene que aplicarse sobre una base llevada hasta la `0004`");
let declaradas = columnas_completas_de(&pool, "backup_jobs").await;
let nombre = declaradas
.iter()
.find(|columna| columna.nombre == "name")
.unwrap_or_else(|| panic!("la `0005` no dejó la columna `name`: {declaradas:?}"));
assert_eq!(nombre.tipo, "TEXT", "`name` tiene que ser de tipo TEXT");
assert!(!nombre.obligatoria, "`name` tiene que admitir NULL");
assert_eq!(
nombre.por_defecto, None,
"`name` NO tiene valor por defecto"
);
let tareas: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM backup_jobs")
.fetch_one(&pool)
.await
.expect("la tabla tiene que poder contarse");
assert_eq!(
tareas, 2,
"la migración se llevó tareas por delante: quedaron {tareas}"
);
let sin_nombre: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM backup_jobs WHERE name IS NULL")
.fetch_one(&pool)
.await
.expect("la columna `name` tiene que poder consultarse");
assert_eq!(
sin_nombre, 2,
"las tareas anteriores al change tienen que quedar con `name` a NULL"
);
pool.close().await;
}
#[tokio::test]
async fn los_enumerados_los_garantiza_un_check_de_la_base() {
let (_directorio, pool) = base_migrada().await;
sqlx::query(
"INSERT INTO backup_targets (id, name, target_type, host_path) \
VALUES ('t-ok', 'carpeta', 'directory', '/srv/datos')",
)
.execute(&pool)
.await
.expect("un target de tipo 'directory' debe entrar");
let fuera_de_enum = sqlx::query(
"INSERT INTO backup_targets (id, name, target_type) VALUES ('t-mal', 'x', 'mysql')",
)
.execute(&pool)
.await
.expect_err("'mysql' no está en el enumerado de target_type");
assert!(
fuera_de_enum
.to_string()
.contains("CHECK constraint failed"),
"el rechazo debe venir del CHECK de la base, no de otra cosa: {fuera_de_enum}"
);
sqlx::query(
"INSERT INTO backup_targets (id, name, target_type, container_name, db_path) \
VALUES ('t-sqlite', 'fichero', 'sqlite', 'ctr-1', '/data/kesto.db')",
)
.execute(&pool)
.await
.expect("'sqlite' tiene que estar en el enumerado de target_type");
let sin_el_tres = sqlx::query(
"INSERT INTO backup_targets (id, name, target_type) VALUES ('t-sqlite3', 'x', 'sqlite3')",
)
.execute(&pool)
.await
.expect_err("'sqlite3' es el binario, no un valor del enumerado de target_type");
assert!(
sin_el_tres.to_string().contains("CHECK constraint failed"),
"el rechazo debe venir del CHECK de la base, no de otra cosa: {sin_el_tres}"
);
sqlx::query(
"INSERT INTO repositories (id, name, endpoint, bucket, region, access_key_id, \
secret_access_key, passphrase) VALUES ('r-ok', 'repo', 'e', 'b', 'r', 'a', 's', 'p')",
)
.execute(&pool)
.await
.expect("un repositorio debe entrar");
sqlx::query(
"INSERT INTO backup_jobs (id, target_id, repository_id, cron_expression) \
VALUES ('j-ok', 't-ok', 'r-ok', '0 3 * * *')",
)
.execute(&pool)
.await
.expect("una tarea debe entrar");
sqlx::query(
"INSERT INTO job_executions (id, job_id, status, started_at) \
VALUES ('e-ok', 'j-ok', 'pending', '2026-01-01T00:00:00Z')",
)
.execute(&pool)
.await
.expect("una ejecución con estado 'pending' debe entrar");
let estado_fuera_de_enum = sqlx::query(
"INSERT INTO job_executions (id, job_id, status, started_at) \
VALUES ('e-mal', 'j-ok', 'cancelled', '2026-01-01T00:00:00Z')",
)
.execute(&pool)
.await
.expect_err("'cancelled' no está en el enumerado de status");
assert!(
estado_fuera_de_enum
.to_string()
.contains("CHECK constraint failed"),
"el rechazo debe venir del CHECK de la base, no de otra cosa: {estado_fuera_de_enum}"
);
pool.close().await;
}
#[tokio::test]
async fn los_cuatro_indices_anadidos_al_srs_existen() {
let (_directorio, pool) = base_migrada().await;
let declarados = sqlx::query_scalar::<_, String>(
"SELECT name FROM sqlite_master WHERE type = 'index' AND name LIKE 'idx_%' ORDER BY name",
)
.fetch_all(&pool)
.await
.expect("la consulta de índices no puede fallar");
for esperado in [
"idx_backup_jobs_target_id",
"idx_backup_jobs_repository_id",
"idx_job_executions_job_id",
"idx_job_executions_status_started_at",
] {
assert!(
declarados.iter().any(|indice| indice == esperado),
"falta el índice {esperado}: {declarados:?}"
);
}
pool.close().await;
}
#[tokio::test]
async fn las_ejecuciones_no_tienen_columna_de_repositorio() {
let (_directorio, pool) = base_migrada().await;
let columnas = columnas_de(&pool, "job_executions").await;
assert!(
!columnas.iter().any(|columna| columna == "repository_id"),
"job_executions no debe llevar repository_id: {columnas:?}"
);
pool.close().await;
}
#[tokio::test]
async fn el_borrado_de_una_fila_padre_no_recorre_la_tabla_hija() {
let (_directorio, pool) = base_migrada().await;
let plan_target = plan_de(
&pool,
"EXPLAIN QUERY PLAN DELETE FROM backup_targets WHERE id = 'x'",
)
.await;
assert!(
plan_target.contains("idx_backup_jobs_target_id"),
"el plan del borrado de un target no usa el índice de la clave foránea: {plan_target}"
);
assert!(
!plan_target.contains("SCAN backup_jobs"),
"el borrado en cascada recorre backup_jobs entera: {plan_target}"
);
let plan_repositorio = plan_de(
&pool,
"EXPLAIN QUERY PLAN DELETE FROM repositories WHERE id = 'x'",
)
.await;
assert!(
plan_repositorio.contains("idx_backup_jobs_repository_id"),
"el plan del borrado de un repositorio no usa el índice de la clave foránea: \
{plan_repositorio}"
);
assert!(
!plan_repositorio.contains("SCAN backup_jobs"),
"el borrado en cascada recorre backup_jobs entera: {plan_repositorio}"
);
let plan_tarea = plan_de(
&pool,
"EXPLAIN QUERY PLAN DELETE FROM backup_jobs WHERE id = 'x'",
)
.await;
assert!(
plan_tarea.contains("idx_job_executions_job_id"),
"el plan del borrado de una tarea no usa el índice de la clave foránea: {plan_tarea}"
);
assert!(
!plan_tarea.contains("SCAN job_executions"),
"el borrado en cascada recorre job_executions entera: {plan_tarea}"
);
pool.close().await;
}
#[tokio::test]
async fn el_listado_de_ejecuciones_usa_su_indice() {
let (_directorio, pool) = base_migrada().await;
let plan = plan_de(
&pool,
"EXPLAIN QUERY PLAN SELECT id, status FROM job_executions \
WHERE status = 'pending' ORDER BY started_at",
)
.await;
assert!(
plan.contains("idx_job_executions_status_started_at"),
"el plan no usa el índice del listado: {plan}"
);
assert!(
!plan.contains("SCAN job_executions"),
"el listado recorre la tabla entera: {plan}"
);
pool.close().await;
}
#[tokio::test]
async fn el_filtro_por_repositorio_usa_los_indices_de_las_claves_foraneas() {
let (_directorio, pool) = base_migrada().await;
let plan = plan_de(
&pool,
"EXPLAIN QUERY PLAN SELECT e.id FROM job_executions e \
JOIN backup_jobs j ON j.id = e.job_id WHERE j.repository_id = 'x'",
)
.await;
assert!(
plan.contains("idx_backup_jobs_repository_id"),
"el plan no usa el índice del repositorio: {plan}"
);
assert!(
plan.contains("idx_job_executions_job_id"),
"el plan no usa el índice de la tarea en las ejecuciones: {plan}"
);
assert!(
!plan.contains("SCAN job_executions"),
"el plan recorre las ejecuciones enteras: {plan}"
);
assert!(
!plan.contains("SCAN backup_jobs"),
"el plan recorre las tareas enteras: {plan}"
);
pool.close().await;
}
#[tokio::test]
async fn la_migracion_0003_amplia_el_enum_sin_perder_filas_ni_indices() {
let (_directorio, pool) = base_hasta(2).await;
let filas = [
(
"t-maria",
"maria",
"mariadb",
Some("maria-1"),
Some("u-maria"),
Some("p-maria"),
Some("d-maria"),
None,
),
(
"t-postgre",
"postgre",
"postgresql",
Some("pg-1"),
Some("u-pg"),
Some("p-pg"),
None,
None,
),
(
"t-carpeta",
"carpeta",
"directory",
None,
None,
None,
None,
Some("/srv/datos"),
),
];
for (id, nombre, tipo, contenedor, usuario, contrasena, base, ruta) in filas {
sqlx::query(
"INSERT INTO backup_targets \
(id, name, target_type, container_name, db_user, db_password, db_name, host_path) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
)
.bind(id)
.bind(nombre)
.bind(tipo)
.bind(contenedor)
.bind(usuario)
.bind(contrasena)
.bind(base)
.bind(ruta)
.execute(&pool)
.await
.expect("los targets previos a la migración deben entrar");
}
sqlx::query(
"INSERT INTO repositories (id, name, endpoint, bucket, region, access_key_id, \
secret_access_key, passphrase) VALUES ('r-1', 'repo', 'e', 'b', 'r', 'a', 's', 'p')",
)
.execute(&pool)
.await
.expect("el repositorio debe entrar");
sqlx::query(
"INSERT INTO backup_jobs (id, target_id, repository_id, cron_expression) \
VALUES ('j-1', 't-maria', 'r-1', '0 3 * * *')",
)
.execute(&pool)
.await
.expect("la tarea sobre el target previo debe entrar");
sqlx::query(
"INSERT INTO job_executions (id, job_id, status, started_at) \
VALUES ('e-1', 'j-1', 'success', '2026-01-01T00:00:00Z')",
)
.execute(&pool)
.await
.expect("la ejecución de esa tarea debe entrar");
super::aplicar(&pool, &super::migrador())
.await
.expect("la 0003 debe aplicarse sobre una base que está en la 0002");
let leidas: Vec<_> = sqlx::query(
"SELECT id, name, target_type, container_name, db_user, db_password, db_name, \
host_path, db_path FROM backup_targets ORDER BY id",
)
.fetch_all(&pool)
.await
.expect("las filas de backup_targets deben poder releerse")
.iter()
.map(|fila| {
(
fila.try_get::<String, _>("id").expect("id es texto"),
fila.try_get::<String, _>("name").expect("name es texto"),
fila.try_get::<String, _>("target_type")
.expect("target_type es texto"),
fila.try_get::<Option<String>, _>("container_name")
.expect("container_name es texto o nulo"),
fila.try_get::<Option<String>, _>("db_user")
.expect("db_user es texto o nulo"),
fila.try_get::<Option<String>, _>("db_password")
.expect("db_password es texto o nulo"),
fila.try_get::<Option<String>, _>("db_name")
.expect("db_name es texto o nulo"),
fila.try_get::<Option<String>, _>("host_path")
.expect("host_path es texto o nulo"),
fila.try_get::<Option<String>, _>("db_path")
.expect("db_path es texto o nulo"),
)
})
.collect();
let esperadas = [
(
"t-carpeta",
"carpeta",
"directory",
None,
None,
None,
None,
Some("/srv/datos"),
None,
),
(
"t-maria",
"maria",
"mariadb",
Some("maria-1"),
Some("u-maria"),
Some("p-maria"),
Some("d-maria"),
None,
None,
),
(
"t-postgre",
"postgre",
"postgresql",
Some("pg-1"),
Some("u-pg"),
Some("p-pg"),
None,
None,
None,
),
]
.map(
|(id, nombre, tipo, contenedor, usuario, contrasena, base, ruta, fichero)| {
(
id.to_owned(),
nombre.to_owned(),
tipo.to_owned(),
contenedor.map(str::to_owned),
usuario.map(str::to_owned),
contrasena.map(str::to_owned),
base.map(str::to_owned),
ruta.map(str::to_owned),
fichero.map(str::to_owned),
)
},
);
assert_eq!(
leidas,
esperadas.to_vec(),
"la reconstrucción de backup_targets perdió o mezcló filas"
);
for (consulta, esperado) in [
("SELECT count(*) FROM backup_jobs", "la tarea"),
("SELECT count(*) FROM job_executions", "la ejecución"),
] {
let cuantas = sqlx::query_scalar::<_, i64>(consulta)
.fetch_one(&pool)
.await
.expect("el recuento no puede fallar");
assert_eq!(
cuantas, 1,
"la cascada se llevó {esperado} por delante durante la reconstrucción"
);
}
sqlx::query(
"INSERT INTO backup_targets (id, name, target_type, container_name, db_path) \
VALUES ('t-sqlite', 'fichero', 'sqlite', 'ctr-1', '/data/kesto.db')",
)
.execute(&pool)
.await
.expect("tras la 0003, 'sqlite' tiene que entrar");
let sin_el_tres = sqlx::query(
"INSERT INTO backup_targets (id, name, target_type) VALUES ('t-sqlite3', 'x', 'sqlite3')",
)
.execute(&pool)
.await
.expect_err("'sqlite3' es el binario, no un valor del enumerado");
assert!(
sin_el_tres.to_string().contains("CHECK constraint failed"),
"el rechazo debe venir del CHECK de la base: {sin_el_tres}"
);
let con_fichero = sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM backup_targets WHERE target_type = 'sqlite' AND db_path IS NULL",
)
.fetch_one(&pool)
.await
.expect("el recuento no puede fallar");
assert_eq!(con_fichero, 0, "el target sqlite quedó sin db_path");
let apuntadas = sqlx::query_scalar::<_, String>(
"SELECT \"table\" FROM pragma_foreign_key_list('backup_jobs')",
)
.fetch_all(&pool)
.await
.expect("las claves foráneas de la tarea deben poder leerse");
assert!(
apuntadas.iter().any(|tabla| tabla == "backup_targets"),
"la clave foránea de la tarea no apunta a backup_targets: {apuntadas:?}"
);
assert!(
!apuntadas.iter().any(|tabla| tabla.ends_with("_nueva")),
"la clave foránea quedó apuntando a la tabla temporal: {apuntadas:?}"
);
sqlx::query("DELETE FROM backup_targets WHERE id = 't-maria'")
.execute(&pool)
.await
.expect("borrar el target no puede fallar");
let tareas = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM backup_jobs")
.fetch_one(&pool)
.await
.expect("el recuento no puede fallar");
let ejecuciones = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM job_executions")
.fetch_one(&pool)
.await
.expect("el recuento no puede fallar");
assert_eq!(
(tareas, ejecuciones),
(0, 0),
"la cascada del §3 dejó de ocurrir tras reconstruir la tabla"
);
for (tabla, esperados) in [
(
"backup_jobs",
["idx_backup_jobs_target_id", "idx_backup_jobs_repository_id"],
),
(
"job_executions",
[
"idx_job_executions_job_id",
"idx_job_executions_status_started_at",
],
),
] {
let indices = sqlx::query_scalar::<_, String>(
"SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = ?1 \
AND name LIKE 'idx_%' ORDER BY name",
)
.bind(tabla)
.fetch_all(&pool)
.await
.expect("los índices deben poder consultarse");
for esperado in esperados {
assert!(
indices.iter().any(|indice| indice == esperado),
"la reconstrucción se dejó {esperado} de {tabla} por el camino: {indices:?}"
);
}
}
let tablas = sqlx::query_scalar::<_, String>(
"SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name",
)
.fetch_all(&pool)
.await
.expect("las tablas deben poder consultarse");
assert!(
!tablas.iter().any(|tabla| tabla.ends_with("_nueva")),
"la migración dejó la tabla temporal en el esquema: {tablas:?}"
);
pool.close().await;
}
#[tokio::test]
async fn la_migracion_0004_indexa_el_snapshot_id() {
const DE_UNA: &str = "SELECT e.job_id, t.id, t.target_type, t.host_path, t.name \
FROM job_executions e \
JOIN backup_jobs j ON j.id = e.job_id \
JOIN backup_targets t ON t.id = j.target_id \
WHERE e.snapshot_id = ?1 AND j.repository_id = ?2 \
ORDER BY e.started_at DESC, e.id DESC LIMIT 1";
const DE_LOTE: &str = "SELECT e.snapshot_id, e.job_id, t.id, t.target_type, t.host_path, \
t.name FROM job_executions e \
JOIN backup_jobs j ON j.id = e.job_id \
JOIN backup_targets t ON t.id = j.target_id \
WHERE e.snapshot_id IN (SELECT value FROM json_each(?1)) AND j.repository_id = ?2 \
ORDER BY e.started_at DESC, e.id DESC";
const SOLO_EL_FILTRO: &str = "SELECT e.id FROM job_executions e WHERE e.snapshot_id = ?1";
let (directorio, pool) = base_hasta(3).await;
poblar_para_el_plan(&pool).await;
let antes = indices_de_tabla(&pool, "job_executions").await;
assert!(
!antes
.iter()
.any(|indice| indice == "idx_job_executions_snapshot_id"),
"el índice de la 0004 no puede existir en una base en la 0003: {antes:?}"
);
for (nombre, consulta) in [("la de una instantánea", DE_UNA), ("la del lote", DE_LOTE)] {
let plan = plan_de(&pool, &format!("EXPLAIN QUERY PLAN {consulta}")).await;
assert!(
!plan.contains("idx_job_executions_snapshot_id"),
"el plan de {nombre} usa un índice que la migración no ha creado: {plan}"
);
assert!(
!plan.contains("SCAN job_executions"),
"el plan de {nombre} recorre `job_executions` entera cuando debería entrar por la \
clave foránea de la tarea: {plan}"
);
}
let suelto_sin_indice =
plan_de(&pool, &format!("EXPLAIN QUERY PLAN {SOLO_EL_FILTRO}")).await;
assert!(
suelto_sin_indice.contains("SCAN e"),
"sin índice, el filtro por `snapshot_id` debería recorrer la tabla: {suelto_sin_indice}"
);
super::aplicar(&pool, &super::migrador())
.await
.expect("la 0004 debe aplicarse sobre una base llevada hasta la 0003");
let declarados = indices_de_tabla(&pool, "job_executions").await;
assert!(
declarados
.iter()
.any(|indice| indice == "idx_job_executions_snapshot_id"),
"la 0004 se aplicó pero el índice no está declarado: {declarados:?}"
);
let dsn = directorio.dsn();
pool.close().await;
let pool = abrir(&dsn)
.await
.expect("la base migrada debe poder reabrirse");
let de_una = plan_de(&pool, &format!("EXPLAIN QUERY PLAN {DE_UNA}")).await;
assert!(
de_una.contains("idx_job_executions_snapshot_id"),
"el plan de la consulta de una instantánea no usa el índice de su columna: {de_una}"
);
let de_lote = plan_de(&pool, &format!("EXPLAIN QUERY PLAN {DE_LOTE}")).await;
assert!(
!de_lote.contains("idx_job_executions_snapshot_id"),
"sin estadísticas, la consulta del lote no tiene por qué usar el índice: {de_lote}"
);
let suelto = plan_de(&pool, &format!("EXPLAIN QUERY PLAN {SOLO_EL_FILTRO}")).await;
assert!(
suelto.contains("idx_job_executions_snapshot_id"),
"el filtro por `snapshot_id` no usa el índice de su columna: {suelto}"
);
assert!(
!suelto.contains("SCAN e"),
"el filtro por `snapshot_id` sigue recorriendo la tabla entera: {suelto}"
);
sqlx::query("ANALYZE")
.execute(&pool)
.await
.expect("`ANALYZE` sobre la base de prueba no puede fallar");
pool.close().await;
let pool = abrir(&dsn)
.await
.expect("la base con estadísticas debe poder reabrirse");
for (nombre, consulta) in [("la de una instantánea", DE_UNA), ("la del lote", DE_LOTE)] {
let plan = plan_de(&pool, &format!("EXPLAIN QUERY PLAN {consulta}")).await;
assert!(
plan.contains("idx_job_executions_snapshot_id"),
"con estadísticas, el plan de {nombre} debería entrar por el índice nuevo: {plan}"
);
}
super::aplicar(&pool, &super::migrador())
.await
.expect("reaplicar el migrador sobre una base ya migrada no puede fallar");
assert_eq!(
indices_de_tabla(&pool, "job_executions").await,
declarados,
"reaplicar la migración cambió los índices de la tabla"
);
pool.close().await;
}
async fn poblar_para_el_plan(pool: &SqlitePool) {
for sentencia in [
"INSERT INTO repositories (id, name, endpoint, bucket, region, access_key_id, \
secret_access_key, passphrase) \
VALUES ('r-1', 'repo', 'e', 'b', 'r', 'a', 's', 'p')",
"INSERT INTO backup_targets (id, name, target_type, host_path) \
VALUES ('t-1', 'carpeta', 'directory', '/srv/datos')",
"INSERT INTO backup_jobs (id, target_id, repository_id, cron_expression) \
SELECT 'j-'||x, 't-1', 'r-1', '0 3 * * *' FROM (WITH RECURSIVE c(x) AS \
(SELECT 1 UNION ALL SELECT x+1 FROM c WHERE x < 200) SELECT x FROM c)",
"INSERT INTO job_executions (id, job_id, status, snapshot_id, started_at) \
SELECT 'e-'||x, 'j-'||((x % 200)+1), 'success', \
CASE WHEN x % 50 = 0 THEN NULL ELSE 'snap-'||x END, \
datetime('now', '-'||x||' seconds') FROM (WITH RECURSIVE c(x) AS \
(SELECT 1 UNION ALL SELECT x+1 FROM c WHERE x < 2000) SELECT x FROM c)",
] {
sqlx::query(sentencia)
.execute(pool)
.await
.expect("la siembra de la prueba 1.54 no puede fallar");
}
}
async fn indices_de_tabla(pool: &SqlitePool, tabla: &str) -> Vec<String> {
sqlx::query_scalar::<_, String>(
"SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = ?1 \
AND name LIKE 'idx_%' ORDER BY name",
)
.bind(tabla)
.fetch_all(pool)
.await
.expect("los índices deben poder consultarse")
}
}