use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::sync::Arc;
use sqlx::Row;
use sqlx::SqlitePool;
use sqlx::sqlite::SqliteRow;
use time::{OffsetDateTime, UtcOffset};
use uuid::Uuid;
use crate::db::DbError;
use crate::db::crypto::Clave;
use crate::db::entities::target::TipoDeTarget;
use crate::secret::SecretString;
#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)]
#[sqlx(rename_all = "lowercase")]
pub enum Estado {
Pending,
Running,
Success,
Failed,
}
impl Estado {
pub fn texto(self) -> &'static str {
match self {
Self::Pending => "pending",
Self::Running => "running",
Self::Success => "success",
Self::Failed => "failed",
}
}
pub fn desde_texto(texto: &str) -> Option<Self> {
match texto {
"pending" => Some(Self::Pending),
"running" => Some(Self::Running),
"success" => Some(Self::Success),
"failed" => Some(Self::Failed),
_ => None,
}
}
}
#[derive(Debug, Clone)]
pub struct Ejecucion {
pub id: Uuid,
pub job_id: Uuid,
pub status: Estado,
pub snapshot_id: Option<String>,
pub bytes_processed: i64,
pub bytes_uploaded: i64,
pub error_message: Option<String>,
pub started_at: OffsetDateTime,
pub finished_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone)]
pub struct AltaDeEjecucion {
pub job_id: Uuid,
pub status: Estado,
pub snapshot_id: Option<String>,
pub bytes_processed: i64,
pub bytes_uploaded: i64,
pub error_message: Option<String>,
pub started_at: OffsetDateTime,
pub finished_at: Option<OffsetDateTime>,
}
pub struct Ejecuciones {
pool: SqlitePool,
clave: Arc<Clave>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FiltroDeEjecuciones {
pub job_id: Option<Uuid>,
pub status: Option<Estado>,
pub limite: i64,
pub desplazamiento: i64,
}
#[derive(Debug, Clone)]
pub struct ObjetivoDeRestauracion {
pub job_id: Uuid,
pub target_id: Uuid,
pub target_type: TipoDeTarget,
pub container_name: Option<String>,
pub db_user: Option<String>,
pub db_password: Option<SecretString>,
pub db_name: Option<String>,
pub host_path: Option<String>,
pub db_path: Option<String>,
pub target_name: String,
}
const SQL_LEER: &str = "SELECT id, job_id, status, snapshot_id, bytes_processed, bytes_uploaded, \
error_message, started_at, finished_at FROM job_executions WHERE id = ?1";
const SQL_LISTAR: &str = "SELECT id, job_id, status, snapshot_id, bytes_processed, bytes_uploaded, \
error_message, started_at, finished_at FROM job_executions ORDER BY started_at, id";
const SQL_LISTAR_FILTRADO: &str = "SELECT id, job_id, status, snapshot_id, bytes_processed, \
bytes_uploaded, error_message, started_at, finished_at FROM job_executions \
WHERE (?1 IS NULL OR job_id = ?1) AND (?2 IS NULL OR status = ?2) \
ORDER BY started_at DESC, id LIMIT ?3 OFFSET ?4";
const SQL_ALTA: &str = "INSERT INTO job_executions \
(id, job_id, status, snapshot_id, bytes_processed, bytes_uploaded, error_message, started_at, \
finished_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) \
RETURNING id, job_id, status, snapshot_id, bytes_processed, bytes_uploaded, error_message, \
started_at, finished_at";
const SQL_ALTA_SI_LIBRE: &str = "INSERT INTO job_executions \
(id, job_id, status, snapshot_id, bytes_processed, bytes_uploaded, error_message, started_at, \
finished_at) \
SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9 \
WHERE NOT EXISTS (SELECT 1 FROM job_executions WHERE job_id = ?2 AND status IN (?10, ?11)) \
RETURNING id, job_id, status, snapshot_id, bytes_processed, bytes_uploaded, error_message, \
started_at, finished_at";
const SQL_ACTUALIZAR: &str = "UPDATE job_executions SET status = ?1, snapshot_id = ?2, \
bytes_processed = ?3, bytes_uploaded = ?4, error_message = ?5, finished_at = ?6 WHERE id = ?7";
const SQL_BORRAR: &str = "DELETE FROM job_executions WHERE id = ?1";
const SQL_OBJETIVO_DE_LA_INSTANTANEA: &str = "SELECT e.job_id, t.id, t.target_type, t.host_path, \
t.name, t.container_name, t.db_user, t.db_password, t.db_name, t.db_path 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 SQL_OBJETIVOS_DE_LAS_INSTANTANEAS: &str = "SELECT e.snapshot_id, e.job_id, t.id, \
t.target_type, t.host_path, t.name, t.container_name, t.db_user, t.db_password, t.db_name, \
t.db_path 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 CONSULTA_LISTA_DE_INSTANTANEAS: &str = "lista JSON de instantáneas del lote";
const CONSULTA_MAPEO: &str = "mapeo de la fila de job_executions";
impl Ejecuciones {
pub fn nuevo(pool: SqlitePool, clave: Arc<Clave>) -> Self {
Ejecuciones { pool, clave }
}
pub async fn alta(&self, alta: AltaDeEjecucion) -> Result<Ejecucion, DbError> {
let id = Uuid::new_v4();
let fila = sqlx::query(SQL_ALTA)
.bind(id.hyphenated().to_string())
.bind(alta.job_id.hyphenated().to_string())
.bind(alta.status)
.bind(&alta.snapshot_id)
.bind(alta.bytes_processed)
.bind(alta.bytes_uploaded)
.bind(&alta.error_message)
.bind(alta.started_at.to_offset(UtcOffset::UTC))
.bind(
alta.finished_at
.map(|fecha| fecha.to_offset(UtcOffset::UTC)),
)
.fetch_one(&self.pool)
.await
.map_err(error_de_consulta(SQL_ALTA))?;
desde_fila(&fila)
}
pub async fn leer(&self, id: &Uuid) -> Result<Option<Ejecucion>, DbError> {
let fila = sqlx::query(SQL_LEER)
.bind(id.hyphenated().to_string())
.fetch_optional(&self.pool)
.await
.map_err(error_de_consulta(SQL_LEER))?;
match fila {
Some(fila) => desde_fila(&fila).map(Some),
None => Ok(None),
}
}
pub async fn listar(&self) -> Result<Vec<Ejecucion>, DbError> {
let filas = sqlx::query(SQL_LISTAR)
.fetch_all(&self.pool)
.await
.map_err(error_de_consulta(SQL_LISTAR))?;
filas.iter().map(desde_fila).collect()
}
pub async fn listar_filtrado(
&self,
filtro: &FiltroDeEjecuciones,
) -> Result<Vec<Ejecucion>, DbError> {
let filas = sqlx::query(SQL_LISTAR_FILTRADO)
.bind(filtro.job_id.map(|id| id.hyphenated().to_string()))
.bind(filtro.status.map(Estado::texto))
.bind(filtro.limite)
.bind(filtro.desplazamiento)
.fetch_all(&self.pool)
.await
.map_err(error_de_consulta(SQL_LISTAR_FILTRADO))?;
filas.iter().map(desde_fila).collect()
}
pub async fn objetivo_de_la_instantanea(
&self,
snapshot_id: &str,
repository_id: &Uuid,
) -> Result<Option<ObjetivoDeRestauracion>, DbError> {
let fila = sqlx::query(SQL_OBJETIVO_DE_LA_INSTANTANEA)
.bind(snapshot_id)
.bind(repository_id.hyphenated().to_string())
.fetch_optional(&self.pool)
.await
.map_err(error_de_consulta(SQL_OBJETIVO_DE_LA_INSTANTANEA))?;
fila.as_ref()
.map(|fila| objetivo_desde_fila(fila, &self.clave))
.transpose()
}
pub async fn objetivos_de_las_instantaneas(
&self,
snapshot_ids: &[String],
repository_id: &Uuid,
) -> Result<HashMap<String, ObjetivoDeRestauracion>, DbError> {
if snapshot_ids.is_empty() {
return Ok(HashMap::new());
}
let lista = serde_json::to_string(snapshot_ids).map_err(|causa| DbError::Consulta {
consulta: CONSULTA_LISTA_DE_INSTANTANEAS,
causa: sqlx::Error::Encode(Box::new(causa)),
})?;
let filas = sqlx::query(SQL_OBJETIVOS_DE_LAS_INSTANTANEAS)
.bind(lista)
.bind(repository_id.hyphenated().to_string())
.fetch_all(&self.pool)
.await
.map_err(error_de_consulta(SQL_OBJETIVOS_DE_LAS_INSTANTANEAS))?;
let mut objetivos = HashMap::with_capacity(snapshot_ids.len());
for fila in &filas {
let instantanea = columna::<String>(fila, "snapshot_id")?;
if let Entry::Vacant(hueco) = objetivos.entry(instantanea) {
hueco.insert(objetivo_desde_fila(fila, &self.clave)?);
}
}
Ok(objetivos)
}
pub async fn actualizar(&self, ejecucion: &Ejecucion) -> Result<bool, DbError> {
let resultado = sqlx::query(SQL_ACTUALIZAR)
.bind(ejecucion.status)
.bind(&ejecucion.snapshot_id)
.bind(ejecucion.bytes_processed)
.bind(ejecucion.bytes_uploaded)
.bind(&ejecucion.error_message)
.bind(
ejecucion
.finished_at
.map(|fecha| fecha.to_offset(UtcOffset::UTC)),
)
.bind(ejecucion.id.hyphenated().to_string())
.execute(&self.pool)
.await
.map_err(error_de_consulta(SQL_ACTUALIZAR))?;
Ok(resultado.rows_affected() == 1)
}
pub async fn alta_si_no_hay_ninguna_en_curso(
&self,
alta: AltaDeEjecucion,
) -> Result<Option<Ejecucion>, DbError> {
let id = Uuid::new_v4();
let fila = sqlx::query(SQL_ALTA_SI_LIBRE)
.bind(id.hyphenated().to_string())
.bind(alta.job_id.hyphenated().to_string())
.bind(alta.status)
.bind(&alta.snapshot_id)
.bind(alta.bytes_processed)
.bind(alta.bytes_uploaded)
.bind(&alta.error_message)
.bind(alta.started_at.to_offset(UtcOffset::UTC))
.bind(
alta.finished_at
.map(|fecha| fecha.to_offset(UtcOffset::UTC)),
)
.bind(Estado::Pending.texto())
.bind(Estado::Running.texto())
.fetch_optional(&self.pool)
.await
.map_err(error_de_consulta(SQL_ALTA_SI_LIBRE))?;
match fila {
Some(fila) => desde_fila(&fila).map(Some),
None => Ok(None),
}
}
pub async fn borrar(&self, id: &Uuid) -> Result<bool, DbError> {
let resultado = sqlx::query(SQL_BORRAR)
.bind(id.hyphenated().to_string())
.execute(&self.pool)
.await
.map_err(error_de_consulta(SQL_BORRAR))?;
Ok(resultado.rows_affected() == 1)
}
}
fn error_de_consulta(consulta: &'static str) -> impl FnOnce(sqlx::Error) -> DbError {
move |causa| DbError::Consulta { consulta, causa }
}
fn columna<'fila, T>(fila: &'fila SqliteRow, nombre: &str) -> Result<T, DbError>
where
T: sqlx::Decode<'fila, sqlx::Sqlite> + sqlx::Type<sqlx::Sqlite>,
{
fila.try_get::<T, _>(nombre)
.map_err(|causa| DbError::Consulta {
consulta: CONSULTA_MAPEO,
causa,
})
}
fn objetivo_desde_fila(fila: &SqliteRow, clave: &Clave) -> Result<ObjetivoDeRestauracion, DbError> {
let job_id = columna::<String>(fila, "job_id")?;
let job_id = Uuid::parse_str(&job_id).map_err(|causa| DbError::Consulta {
consulta: CONSULTA_MAPEO,
causa: sqlx::Error::Decode(Box::new(causa)),
})?;
let target_id = columna::<String>(fila, "id")?;
let target_id = Uuid::parse_str(&target_id).map_err(|causa| DbError::Consulta {
consulta: CONSULTA_MAPEO,
causa: sqlx::Error::Decode(Box::new(causa)),
})?;
let db_password = match columna::<Option<String>>(fila, "db_password")? {
Some(cifrado) => Some(clave.descifrar(&cifrado)?),
None => None,
};
Ok(ObjetivoDeRestauracion {
job_id,
target_id,
target_type: columna(fila, "target_type")?,
container_name: columna(fila, "container_name")?,
db_user: columna(fila, "db_user")?,
db_password,
db_name: columna(fila, "db_name")?,
host_path: columna(fila, "host_path")?,
db_path: columna(fila, "db_path")?,
target_name: columna(fila, "name")?,
})
}
fn desde_fila(fila: &SqliteRow) -> Result<Ejecucion, DbError> {
let id = columna::<String>(fila, "id")?;
let id = Uuid::parse_str(&id).map_err(|causa| DbError::Consulta {
consulta: CONSULTA_MAPEO,
causa: sqlx::Error::Decode(Box::new(causa)),
})?;
let job_id = columna::<String>(fila, "job_id")?;
let job_id = Uuid::parse_str(&job_id).map_err(|causa| DbError::Consulta {
consulta: CONSULTA_MAPEO,
causa: sqlx::Error::Decode(Box::new(causa)),
})?;
Ok(Ejecucion {
id,
job_id,
status: columna(fila, "status")?,
snapshot_id: columna(fila, "snapshot_id")?,
bytes_processed: columna(fila, "bytes_processed")?,
bytes_uploaded: columna(fila, "bytes_uploaded")?,
error_message: columna(fila, "error_message")?,
started_at: columna(fila, "started_at")?,
finished_at: columna(fila, "finished_at")?,
})
}
#[cfg(test)]
mod tests {
use futures_util::future::join_all;
use sqlx::SqlitePool;
use time::macros::datetime;
use time::{Duration, OffsetDateTime};
use uuid::Uuid;
use super::{AltaDeEjecucion, Ejecuciones, Estado, FiltroDeEjecuciones};
use crate::db::DbError;
use crate::db::entities::job::{AltaDeTarea, Tareas};
use crate::db::entities::repository::{AltaDeRepositorio, Repositorios};
use crate::db::entities::target::{AltaDeTarget, Targets, TipoDeTarget};
use crate::secret::SecretString;
use crate::test_support::{DirectorioDePrueba, base_con_clave, clave_de_prueba};
const SNAPSHOT: &str = "snapshot-distinguible-000000000000000000000001";
const OTRA_INSTANTANEA: &str = "snapshot-distinguible-000000000000000000000002";
fn secreto(valor: &str) -> SecretString {
SecretString::new(valor.to_owned())
}
fn alta_repositorio(nombre: &str) -> AltaDeRepositorio {
AltaDeRepositorio {
name: nombre.to_owned(),
endpoint: format!("https://s3.example.test/{nombre}"),
bucket: format!("cubo-{nombre}"),
region: "eu-west-1".to_owned(),
access_key_id: "AKIAIOSFODNN7EXAMPLE".to_owned(),
secret_access_key: secreto("AKIAIOSFODNN7EXAMPLE-de-ejecucion"),
passphrase: secreto("passphrase-distinguible-de-ejecucion"),
}
}
fn alta_target(nombre: &str) -> AltaDeTarget {
AltaDeTarget {
name: nombre.to_owned(),
target_type: TipoDeTarget::Directory,
container_name: None,
db_user: None,
db_password: None,
db_name: None,
host_path: Some(format!("/srv/{nombre}")),
db_path: None,
}
}
fn alta_ejecucion(job_id: &Uuid) -> AltaDeEjecucion {
let started_at = OffsetDateTime::now_utc();
AltaDeEjecucion {
job_id: *job_id,
status: Estado::Success,
snapshot_id: Some(SNAPSHOT.to_owned()),
bytes_processed: 1024,
bytes_uploaded: 512,
error_message: None,
started_at,
finished_at: Some(started_at + Duration::seconds(30)),
}
}
fn alta_con(instantanea: &str, job_id: &Uuid, started_at: OffsetDateTime) -> AltaDeEjecucion {
AltaDeEjecucion {
job_id: *job_id,
status: Estado::Success,
snapshot_id: Some(instantanea.to_owned()),
bytes_processed: 1024,
bytes_uploaded: 512,
error_message: None,
started_at,
finished_at: Some(started_at + Duration::seconds(30)),
}
}
fn alta_encolada(job_id: &Uuid) -> AltaDeEjecucion {
AltaDeEjecucion {
status: Estado::Pending,
snapshot_id: None,
error_message: None,
finished_at: None,
..alta_ejecucion(job_id)
}
}
async fn base_con_ejecucion_posible() -> (DirectorioDePrueba, SqlitePool, Ejecuciones, Uuid) {
let (directorio, pool, clave) = base_con_clave().await;
let repositorio = Repositorios::nuevo(pool.clone(), clave.clone())
.alta(alta_repositorio("repo-de-ejecucion"))
.await
.expect("el alta del repositorio no puede fallar");
let target = Targets::nuevo(pool.clone(), clave)
.alta(alta_target("target-de-ejecucion"))
.await
.expect("el alta del target no puede fallar");
let tarea = Tareas::nuevo(pool.clone())
.alta(AltaDeTarea {
target_id: target.id,
repository_id: repositorio.id,
cron_expression: "0 3 * * *".to_owned(),
keep_daily: 7,
keep_weekly: 4,
keep_monthly: 12,
enabled: true,
name: None,
})
.await
.expect("el alta de la tarea no puede fallar");
let ejecuciones = Ejecuciones::nuevo(pool.clone(), clave_de_prueba());
(directorio, pool, ejecuciones, tarea.id)
}
async fn filas_de_la_tarea(pool: &SqlitePool, job_id: &Uuid) -> i64 {
sqlx::query_scalar("SELECT COUNT(*) FROM job_executions WHERE job_id = ?1")
.bind(job_id.hyphenated().to_string())
.fetch_one(pool)
.await
.expect("el recuento de filas debe poder leerse")
}
async fn tarea_hermana(pool: &SqlitePool, job_id: &Uuid) -> Uuid {
let tareas = Tareas::nuevo(pool.clone());
let original = tareas
.leer(job_id)
.await
.expect("la lectura de la tarea no puede fallar")
.expect("la tarea de la base de prueba tiene que estar");
tareas
.alta(AltaDeTarea {
target_id: original.target_id,
repository_id: original.repository_id,
cron_expression: "0 4 * * *".to_owned(),
keep_daily: 7,
keep_weekly: 4,
keep_monthly: 12,
enabled: true,
name: None,
})
.await
.expect("el alta de la tarea hermana no puede fallar")
.id
}
#[tokio::test]
async fn el_ciclo_de_una_ejecucion_y_su_estado_guardado_como_texto() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let dado_de_alta = ejecuciones
.alta(alta_ejecucion(&job_id))
.await
.expect("el alta no puede fallar");
let leido = ejecuciones
.leer(&dado_de_alta.id)
.await
.expect("la lectura no puede fallar")
.expect("la fila recién dada de alta debe estar");
assert_eq!(leido.id, dado_de_alta.id, "el id leído no es el del alta");
assert_eq!(leido.job_id, job_id);
assert_eq!(leido.status, Estado::Success);
assert_eq!(leido.snapshot_id.as_deref(), Some(SNAPSHOT));
assert_eq!(leido.bytes_processed, 1024);
assert_eq!(leido.bytes_uploaded, 512);
assert_eq!(leido.error_message, None);
assert_eq!(leido.started_at, dado_de_alta.started_at);
assert_eq!(leido.finished_at, dado_de_alta.finished_at);
assert!(
leido.finished_at.expect("terminó") > leido.started_at,
"el final debe ser posterior al inicio"
);
let estado_almacenado: String =
sqlx::query_scalar("SELECT status FROM job_executions WHERE id = ?1")
.bind(dado_de_alta.id.hyphenated().to_string())
.fetch_one(&pool)
.await
.expect("el estado debe poder leerse");
assert_eq!(
estado_almacenado, "success",
"el enumerado no se guardó como texto, sino como {estado_almacenado}"
);
let listado = ejecuciones
.listar()
.await
.expect("el listado no puede fallar");
assert_eq!(listado.len(), 1, "el listado debe traer la ejecución");
let mut cambiada = leido.clone();
cambiada.status = Estado::Failed;
cambiada.error_message = Some("fallo distinguible de la prueba 1-30".to_owned());
assert!(
ejecuciones
.actualizar(&cambiada)
.await
.expect("la actualización no puede fallar"),
"la fila existía, así que la actualización debe decir que sí"
);
let releida = ejecuciones
.leer(&dado_de_alta.id)
.await
.expect("la lectura no puede fallar")
.expect("la fila debe seguir existiendo");
assert_eq!(releida.status, Estado::Failed, "el estado tocado no cambió");
assert_eq!(
releida.error_message.as_deref(),
Some("fallo distinguible de la prueba 1-30"),
"el mensaje de error no se escribió"
);
assert_eq!(
releida.bytes_processed, leido.bytes_processed,
"los contadores no debían cambiar"
);
assert_eq!(
releida.started_at, dado_de_alta.started_at,
"el started_at no debía cambiar"
);
assert!(
ejecuciones
.borrar(&dado_de_alta.id)
.await
.expect("el borrado no puede fallar"),
"la fila existía, así que el borrado debe decir que sí"
);
let tras_borrar = ejecuciones
.leer(&dado_de_alta.id)
.await
.expect("la lectura no puede fallar");
assert!(tras_borrar.is_none(), "la fila debía desaparecer");
}
#[tokio::test]
async fn el_finished_at_nulo_se_lee_como_none() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let en_curso = ejecuciones
.alta(AltaDeEjecucion {
status: Estado::Running,
snapshot_id: None,
finished_at: None,
..alta_ejecucion(&job_id)
})
.await
.expect("el alta no puede fallar");
let leida = ejecuciones
.leer(&en_curso.id)
.await
.expect("la lectura no puede fallar")
.expect("la fila debe estar");
assert_eq!(
leida.finished_at, None,
"un finished_at NULL debe leerse como `None`, no como una fecha inventada"
);
assert_eq!(leida.status, Estado::Running);
assert_eq!(
leida.snapshot_id, None,
"una ejecución sin snapshot debe leerse sin él"
);
let es_nulo: i64 =
sqlx::query_scalar("SELECT finished_at IS NULL FROM job_executions WHERE id = ?1")
.bind(en_curso.id.hyphenated().to_string())
.fetch_one(&pool)
.await
.expect("el `IS NULL` debe poder leerse");
assert_eq!(es_nulo, 1, "la columna no es NULL, sino otra cosa");
}
#[tokio::test]
async fn una_fecha_con_desplazamiento_vuelve_en_utc() {
let (_directorio, _pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let inicio = datetime!(2026-03-14 09:26:53 +2:30);
let fin = inicio + Duration::minutes(7);
assert!(
!inicio.offset().is_utc(),
"el aparato de la prueba no usó un desplazamiento distinto de cero"
);
let dado_de_alta = ejecuciones
.alta(AltaDeEjecucion {
started_at: inicio,
finished_at: Some(fin),
..alta_ejecucion(&job_id)
})
.await
.expect("el alta no puede fallar");
let leido = ejecuciones
.leer(&dado_de_alta.id)
.await
.expect("la lectura no puede fallar")
.expect("la fila debe estar");
assert!(
leido.started_at.offset().is_utc(),
"el started_at volvió con desplazamiento, no en UTC: {}",
leido.started_at
);
assert_eq!(
leido.started_at, inicio,
"el instante guardado no es el de entrada"
);
let fin_leido = leido
.finished_at
.expect("la ejecución se dio de alta terminada");
assert!(
fin_leido.offset().is_utc(),
"el finished_at volvió con desplazamiento, no en UTC: {fin_leido}"
);
assert_eq!(fin_leido, fin, "el instante final no es el de entrada");
}
#[tokio::test]
async fn el_listado_filtrado_acota_y_ordena_por_la_fecha_descendente() {
let (directorio, _pool, ejecuciones, tarea) = base_con_ejecucion_posible().await;
let empate = datetime!(2026-10-25 03:00:00 UTC);
let anterior = empate - Duration::hours(1);
let mut identificadores = Vec::new();
for (estado, fecha, procesados) in [
(Estado::Success, anterior, 100_i64),
(Estado::Failed, empate, 200),
(Estado::Running, empate, 300),
] {
let fila = ejecuciones
.alta(AltaDeEjecucion {
job_id: tarea,
status: estado,
snapshot_id: None,
bytes_processed: procesados,
bytes_uploaded: procesados / 2,
error_message: None,
started_at: fecha,
finished_at: None,
})
.await
.expect("el alta de la ejecución no puede fallar");
identificadores.push(fila.id);
}
let sin_filtros = FiltroDeEjecuciones {
job_id: None,
status: None,
limite: 50,
desplazamiento: 0,
};
let todas = ejecuciones
.listar_filtrado(&sin_filtros)
.await
.expect("el listado filtrado no puede fallar");
assert_eq!(
todas.len(),
3,
"sin filtros tienen que salir las tres filas"
);
assert_eq!(
todas[2].bytes_processed, 100,
"la más antigua tiene que quedar la última, y ahí no hay empate que valga"
);
let mut de_las_empatadas = [todas[0].bytes_processed, todas[1].bytes_processed];
de_las_empatadas.sort();
assert_eq!(
de_las_empatadas,
[200, 300],
"las dos filas que empatan tienen que ir delante de la más antigua"
);
let mut esperadas = [
identificadores[1].hyphenated().to_string(),
identificadores[2].hyphenated().to_string(),
];
esperadas.sort();
assert_eq!(
todas[0].id.hyphenated().to_string(),
esperadas[0],
"con la misma fecha, el desempate es el identificador"
);
assert_eq!(
todas[1].id.hyphenated().to_string(),
esperadas[1],
"y el de la segunda también"
);
let solo_fallidas = ejecuciones
.listar_filtrado(&FiltroDeEjecuciones {
job_id: None,
status: Some(Estado::Failed),
limite: 50,
desplazamiento: 0,
})
.await
.expect("el listado filtrado no puede fallar");
assert_eq!(
solo_fallidas.len(),
1,
"el filtro por estado tiene que acotar"
);
let de_otra_tarea = ejecuciones
.listar_filtrado(&FiltroDeEjecuciones {
job_id: Some(Uuid::new_v4()),
status: None,
limite: 50,
desplazamiento: 0,
})
.await
.expect("el listado filtrado no puede fallar");
assert!(
de_otra_tarea.is_empty(),
"un filtro por una tarea sin ejecuciones no puede devolver filas"
);
let segunda = ejecuciones
.listar_filtrado(&FiltroDeEjecuciones {
job_id: None,
status: None,
limite: 1,
desplazamiento: 1,
})
.await
.expect("el listado filtrado no puede fallar");
assert_eq!(
segunda.len(),
1,
"la ventana tiene que devolver una sola fila"
);
assert_eq!(
segunda[0].id, todas[1].id,
"saltar una tiene que dar la segunda del orden"
);
drop(directorio);
}
#[tokio::test]
async fn con_la_tarea_libre_la_guarda_inserta_y_devuelve_la_fila() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let creada = ejecuciones
.alta_si_no_hay_ninguna_en_curso(alta_encolada(&job_id))
.await
.expect("el alta no puede fallar")
.expect("con la tarea libre la guarda tiene que dejar pasar el alta");
assert_eq!(
creada.job_id, job_id,
"la fila devuelta no es de la tarea que se pidió"
);
assert_eq!(
creada.status,
Estado::Pending,
"el estado devuelto no es el del alta"
);
assert_eq!(
filas_de_la_tarea(&pool, &job_id).await,
1,
"el alta tiene que dejar exactamente una fila"
);
let leida = ejecuciones
.leer(&creada.id)
.await
.expect("la lectura no puede fallar")
.expect("la fila que devolvió la guarda tiene que estar en la tabla");
assert_eq!(
leida.id, creada.id,
"la fila devuelta por la guarda no es la que quedó en la tabla"
);
}
#[tokio::test]
async fn una_ejecucion_pending_en_curso_rechaza_el_alta_sin_crear_fila() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let primera = ejecuciones
.alta(alta_encolada(&job_id))
.await
.expect("el alta previa no puede fallar");
let segunda = ejecuciones
.alta_si_no_hay_ninguna_en_curso(alta_encolada(&job_id))
.await
.expect("la guarda rechaza el alta **sin fallar**: `None` no es un error");
assert!(
segunda.is_none(),
"con una `pending` en curso la guarda tiene que devolver `None`"
);
assert_eq!(
filas_de_la_tarea(&pool, &job_id).await,
1,
"la guarda dijo que no, pero dejó una segunda fila en la tabla"
);
let restante = ejecuciones
.listar()
.await
.expect("el listado no puede fallar");
assert_eq!(
restante.len(),
1,
"tiene que quedar solo la ejecución previa"
);
assert_eq!(
restante[0].id, primera.id,
"la fila que queda no es la primera: la guarda insertó de todos modos"
);
}
#[tokio::test]
async fn una_ejecucion_running_en_curso_rechaza_el_alta_sin_crear_fila() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let en_curso = ejecuciones
.alta(AltaDeEjecucion {
status: Estado::Running,
..alta_encolada(&job_id)
})
.await
.expect("el alta previa no puede fallar");
let segunda = ejecuciones
.alta_si_no_hay_ninguna_en_curso(alta_encolada(&job_id))
.await
.expect("la guarda rechaza el alta **sin fallar**: `None` no es un error");
assert!(
segunda.is_none(),
"con una `running` en curso la guarda tiene que devolver `None`"
);
assert_eq!(
filas_de_la_tarea(&pool, &job_id).await,
1,
"la guarda dijo que no, pero dejó una segunda fila en la tabla"
);
assert_eq!(
en_curso.status,
Estado::Running,
"el aparato de la prueba no dejó la ejecución en curso"
);
}
#[tokio::test]
async fn el_historico_terminado_no_bloquea_una_copia_nueva() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let otra_tarea = tarea_hermana(&pool, &job_id).await;
for (tarea, estado) in [(&job_id, Estado::Success), (&otra_tarea, Estado::Failed)] {
ejecuciones
.alta(AltaDeEjecucion {
status: estado,
..alta_ejecucion(tarea)
})
.await
.expect("el alta del histórico no puede fallar");
let creada = ejecuciones
.alta_si_no_hay_ninguna_en_curso(alta_encolada(tarea))
.await
.expect("el alta no puede fallar");
assert!(
creada.is_some(),
"una ejecución `{estado:?}` de ayer no puede dejar la tarea clavada para siempre"
);
assert_eq!(
filas_de_la_tarea(&pool, tarea).await,
2,
"tienen que quedar la del histórico y la nueva"
);
}
}
#[tokio::test]
async fn seis_altas_simultaneas_dejan_una_sola_fila() {
const SIMULTANEAS: usize = 6;
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let altas = (0..SIMULTANEAS)
.map(|_| ejecuciones.alta_si_no_hay_ninguna_en_curso(alta_encolada(&job_id)));
let mut creadas = Vec::new();
for resultado in join_all(altas).await {
let resultado = resultado.expect(
"ninguna de las altas simultáneas puede fallar: un fallo de base no es «la tarea \
ya estaba en curso»",
);
if let Some(fila) = resultado {
creadas.push(fila);
}
}
assert_eq!(
creadas.len(),
1,
"de {SIMULTANEAS} altas simultáneas sobre la misma tarea salieron {}: \
la guarda no es atómica",
creadas.len()
);
assert_eq!(
filas_de_la_tarea(&pool, &job_id).await,
1,
"la tabla tiene que quedar con **exactamente una** fila de la tarea"
);
}
#[tokio::test]
async fn la_ejecucion_en_curso_de_otra_tarea_no_bloquea_el_alta() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let otra_tarea = tarea_hermana(&pool, &job_id).await;
ejecuciones
.alta(alta_encolada(&job_id))
.await
.expect("el alta previa no puede fallar");
let creada = ejecuciones
.alta_si_no_hay_ninguna_en_curso(alta_encolada(&otra_tarea))
.await
.expect("el alta no puede fallar");
assert!(
creada.is_some(),
"la ejecución en curso de otra tarea no puede bloquear el alta de ésta"
);
assert_eq!(
filas_de_la_tarea(&pool, &job_id).await,
1,
"la tarea que ya estaba en curso no puede llevarse una segunda fila"
);
assert_eq!(
filas_de_la_tarea(&pool, &otra_tarea).await,
1,
"la tarea de al lado tiene que quedarse con su fila"
);
}
#[tokio::test]
async fn una_tarea_que_no_existe_falla_en_vez_de_devolver_none() {
let (_directorio, pool, ejecuciones, _job_id) = base_con_ejecucion_posible().await;
let inexistente = Uuid::new_v4();
let error = ejecuciones
.alta_si_no_hay_ninguna_en_curso(alta_encolada(&inexistente))
.await
.expect_err("una tarea inexistente tiene que fallar, no decir que ya estaba en curso");
assert!(
matches!(error, DbError::Consulta { .. }),
"el fallo tiene que llegar como error de consulta, no como otro: {error}"
);
assert_eq!(
filas_de_la_tarea(&pool, &inexistente).await,
0,
"una ejecución huérfana no puede llegar a existir"
);
}
#[tokio::test]
async fn el_objetivo_de_una_instantanea_sale_del_vinculo_de_su_ejecucion() {
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let tarea = Tareas::nuevo(pool.clone())
.leer(&job_id)
.await
.expect("la lectura de la tarea no puede fallar")
.expect("la tarea de la base de prueba tiene que estar");
assert!(
ejecuciones
.objetivo_de_la_instantanea(SNAPSHOT, &tarea.repository_id)
.await
.expect("la consulta no puede fallar")
.is_none(),
"sin ninguna ejecución que anote la instantánea no hay vínculo que resolver"
);
ejecuciones
.alta(alta_ejecucion(&job_id))
.await
.expect("el alta de la ejecución no puede fallar");
let objetivo = ejecuciones
.objetivo_de_la_instantanea(SNAPSHOT, &tarea.repository_id)
.await
.expect("la consulta no puede fallar")
.expect("la ejecución anotó la instantánea, así que el vínculo existe");
assert_eq!(objetivo.job_id, job_id, "la tarea no es la de la ejecución");
assert_eq!(
objetivo.target_id, tarea.target_id,
"el target no es el de la tarea de la ejecución"
);
assert_eq!(objetivo.target_type, TipoDeTarget::Directory);
assert_eq!(
objetivo.host_path.as_deref(),
Some("/srv/target-de-ejecucion")
);
assert_eq!(objetivo.target_name, "target-de-ejecucion");
assert!(
ejecuciones
.objetivo_de_la_instantanea("otra-instantanea", &tarea.repository_id)
.await
.expect("la consulta no puede fallar")
.is_none(),
"una instantánea que ninguna ejecución anotó no tiene vínculo"
);
assert!(
ejecuciones
.objetivo_de_la_instantanea(SNAPSHOT, &Uuid::new_v4())
.await
.expect("la consulta no puede fallar")
.is_none(),
"el vínculo se acota al repositorio por el que se pregunta: la copia de otro \
repositorio no resuelve aquí"
);
}
#[tokio::test]
async fn el_objetivo_de_la_instantanea_trae_los_campos_del_target_y_el_secreto_descifrado() {
const PASSWORD: &str = "db-password-distinguible-de-la-restauracion";
let (_directorio, pool, clave) = base_con_clave().await;
let repositorio = Repositorios::nuevo(pool.clone(), clave.clone())
.alta(alta_repositorio("repo-de-la-restauracion"))
.await
.expect("el alta del repositorio no puede fallar");
let target = Targets::nuevo(pool.clone(), clave)
.alta(AltaDeTarget {
name: "target-de-base-de-la-restauracion".to_owned(),
target_type: TipoDeTarget::Mariadb,
container_name: Some("contenedor-de-la-base".to_owned()),
db_user: Some("operador".to_owned()),
db_password: Some(secreto(PASSWORD)),
db_name: Some("tienda".to_owned()),
host_path: None,
db_path: None,
})
.await
.expect("el alta del target no puede fallar");
let tarea = Tareas::nuevo(pool.clone())
.alta(AltaDeTarea {
target_id: target.id,
repository_id: repositorio.id,
cron_expression: "0 3 * * *".to_owned(),
keep_daily: 7,
keep_weekly: 4,
keep_monthly: 12,
enabled: true,
name: None,
})
.await
.expect("el alta de la tarea no puede fallar");
let ejecuciones = Ejecuciones::nuevo(pool.clone(), clave_de_prueba());
ejecuciones
.alta(alta_ejecucion(&tarea.id))
.await
.expect("el alta de la ejecución no puede fallar");
let objetivo = ejecuciones
.objetivo_de_la_instantanea(SNAPSHOT, &repositorio.id)
.await
.expect("la consulta no puede fallar")
.expect("la ejecución anotó la instantánea, así que el vínculo existe");
assert_eq!(objetivo.target_id, target.id);
assert_eq!(objetivo.target_type, TipoDeTarget::Mariadb);
assert_eq!(objetivo.target_name, "target-de-base-de-la-restauracion");
assert_eq!(
objetivo.container_name.as_deref(),
Some("contenedor-de-la-base"),
"sin el contenedor no hay dónde ejecutar la restauración"
);
assert_eq!(objetivo.db_user.as_deref(), Some("operador"));
assert_eq!(objetivo.db_name.as_deref(), Some("tienda"));
assert_eq!(
objetivo.db_path, None,
"una base de MariaDB no tiene ruta de fichero dentro del contenedor"
);
assert_eq!(
objetivo.host_path, None,
"un target de base no tiene ruta de host"
);
assert_eq!(
objetivo.db_password.as_ref().map(SecretString::expose),
Some(PASSWORD),
"la contraseña no llegó descifrada"
);
let almacenado: String =
sqlx::query_scalar("SELECT db_password FROM backup_targets WHERE id = ?1")
.bind(target.id.hyphenated().to_string())
.fetch_one(&pool)
.await
.expect("el secreto almacenado debe poder leerse");
assert!(
!almacenado.contains(PASSWORD),
"la contraseña está en claro en la columna: {almacenado}"
);
let lote = ejecuciones
.objetivos_de_las_instantaneas(&[SNAPSHOT.to_owned()], &repositorio.id)
.await
.expect("la consulta del lote no puede fallar");
let del_lote = lote
.get(SNAPSHOT)
.expect("la instantánea anotada tiene que salir en el lote");
assert_eq!(
del_lote.container_name.as_deref(),
Some("contenedor-de-la-base")
);
assert_eq!(del_lote.db_user.as_deref(), Some("operador"));
assert_eq!(del_lote.db_name.as_deref(), Some("tienda"));
assert_eq!(
del_lote.db_password.as_ref().map(SecretString::expose),
Some(PASSWORD),
"el lote no trae la contraseña descifrada"
);
assert!(
!format!("{objetivo:?}").contains(PASSWORD),
"el Debug del objetivo publica la contraseña: {objetivo:?}"
);
}
#[tokio::test]
async fn los_objetivos_de_un_lote_salen_del_vinculo_de_cada_ejecucion() {
const SIN_VINCULO: &str = "instantanea-que-nadie-anoto";
let (_directorio, pool, ejecuciones, job_id) = base_con_ejecucion_posible().await;
let hermana = tarea_hermana(&pool, &job_id).await;
let tarea = Tareas::nuevo(pool.clone())
.leer(&job_id)
.await
.expect("la lectura de la tarea no puede fallar")
.expect("la tarea de la base de prueba tiene que estar");
ejecuciones
.alta(alta_con(
SNAPSHOT,
&job_id,
datetime!(2026-01-01 10:00:00 UTC),
))
.await
.expect("el alta de la ejecución no puede fallar");
ejecuciones
.alta(alta_con(
SNAPSHOT,
&hermana,
datetime!(2026-01-02 10:00:00 UTC),
))
.await
.expect("el alta de la ejecución hermana no puede fallar");
ejecuciones
.alta(alta_con(
OTRA_INSTANTANEA,
&job_id,
datetime!(2026-01-03 10:00:00 UTC),
))
.await
.expect("el alta de la ejecución de la otra instantánea no puede fallar");
let objetivos = ejecuciones
.objetivos_de_las_instantaneas(
&[
SNAPSHOT.to_owned(),
OTRA_INSTANTANEA.to_owned(),
SIN_VINCULO.to_owned(),
],
&tarea.repository_id,
)
.await
.expect("la consulta del lote no puede fallar");
assert_eq!(
objetivos.len(),
2,
"el lote tiene que traer las dos instantáneas con vínculo y ninguna más: {objetivos:?}"
);
let con_vinculo = objetivos
.get(SNAPSHOT)
.expect("la instantánea anotada por dos ejecuciones tiene que salir en el lote");
assert_eq!(
con_vinculo.job_id, hermana,
"de dos ejecuciones de la misma instantánea gana la más reciente, y ganó la primera"
);
assert_eq!(
con_vinculo.target_id, tarea.target_id,
"el target no es el de la tarea de la ejecución"
);
assert_eq!(con_vinculo.target_type, TipoDeTarget::Directory);
assert_eq!(
con_vinculo.host_path.as_deref(),
Some("/srv/target-de-ejecucion")
);
assert_eq!(con_vinculo.target_name, "target-de-ejecucion");
let la_otra = objetivos
.get(OTRA_INSTANTANEA)
.expect("la otra instantánea con vínculo tiene que salir con **su** objetivo");
assert_eq!(
la_otra.job_id, job_id,
"el lote tiene que casar cada instantánea con su propia ejecución, y no devolver el \
objetivo de la primera para todas"
);
assert!(
!objetivos.contains_key(SIN_VINCULO),
"una instantánea que ninguna ejecución anotó no tiene vínculo y no puede salir en el \
mapa: {objetivos:?}"
);
let de_otro_repositorio = ejecuciones
.objetivos_de_las_instantaneas(&[SNAPSHOT.to_owned()], &Uuid::new_v4())
.await
.expect("la consulta del lote no puede fallar");
assert!(
de_otro_repositorio.is_empty(),
"el lote se acota al repositorio por el que se pregunta: la copia de otro repositorio \
no resuelve aquí"
);
let vacio = ejecuciones
.objetivos_de_las_instantaneas(&[], &tarea.repository_id)
.await
.expect("un lote vacío no puede fallar");
assert!(
vacio.is_empty(),
"un lote sin identificadores no tiene nada que resolver y tiene que salir vacío"
);
}
}