mod cola;
mod ejecucion;
pub mod extraccion;
mod hub;
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
use time::OffsetDateTime;
use tokio::runtime::Handle;
use uuid::Uuid;
use crate::db::DbError;
use crate::db::entities::{
AltaDeEjecucion, Ejecucion, Ejecuciones, Estado, Repositorios, Tareas, Targets,
};
use crate::docker::Cliente;
use crate::rustic::{Acceso, BarrasDelMotor, PuestaEnMarcha, Repositorio, ResumenDeSnapshot};
use cola::Cola;
pub use hub::{CAPACIDAD_DEL_HUB, EventoDeEjecucion, Hub};
pub use extraccion::{
CAPACIDAD_DEL_CANAL, ErrorDeExtraccion, INACTIVIDAD_MAXIMA, Objetivo, Resultado, Ruta, ruta,
volcar,
};
const ESPERA_DEL_MOTOR: Duration = Duration::from_secs(300);
const ESPERA_DE_PARADA: Duration = Duration::from_secs(5);
pub type AbridorDeRepositorios =
fn(&Acceso, BarrasDelMotor) -> Result<(Repositorio, PuestaEnMarcha), crate::rustic::Error>;
#[derive(Clone)]
pub struct Ajustes {
pub socket_del_motor: String,
pub espera_del_motor: Duration,
pub espera_de_parada: Duration,
pub abridor: AbridorDeRepositorios,
}
impl Ajustes {
pub fn nuevos(socket: impl Into<String>) -> Self {
Self {
socket_del_motor: socket.into(),
espera_del_motor: ESPERA_DEL_MOTOR,
espera_de_parada: ESPERA_DE_PARADA,
abridor: Repositorio::poner_en_marcha,
}
}
pub fn con_abridor(mut self, abridor: AbridorDeRepositorios) -> Self {
self.abridor = abridor;
self
}
}
impl std::fmt::Debug for Ajustes {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Ajustes")
.field("socket_del_motor", &self.socket_del_motor)
.field("espera_del_motor", &self.espera_del_motor)
.field("espera_de_parada", &self.espera_de_parada)
.finish_non_exhaustive()
}
}
#[derive(Clone)]
pub struct Dependencias {
pub repositorios: Arc<Repositorios>,
pub targets: Arc<Targets>,
pub tareas: Arc<Tareas>,
pub ejecuciones: Arc<Ejecuciones>,
}
impl std::fmt::Debug for Dependencias {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Dependencias").finish_non_exhaustive()
}
}
#[derive(Debug, Error)]
pub enum ErrorDelMotor {
#[error("la tarea `{job_id}` no está: no hay nada que encolar")]
TareaQueNoEsta {
job_id: Uuid,
},
#[error("la tarea `{job_id}` ya tiene una ejecución en curso: no se encola una segunda")]
YaTieneUnaEnCurso {
job_id: Uuid,
},
#[error("el motor está parado: no acepta trabajo nuevo")]
MotorParado,
#[error(
"la cola del repositorio `{repository_id}` está llena: la ejecución no ha podido entrar, y \
se ha apuntado como fallida en vez de dejarla esperando"
)]
ColaDelRepositorioLlena {
repository_id: Uuid,
},
#[error("el repositorio `{repository_id}` no tiene quien atienda su cola: {motivo}")]
SinQuienAtiendaLaCola {
repository_id: Uuid,
motivo: String,
},
#[error("la base no ha podido atender el encolado: {causa}")]
Base {
#[source]
causa: DbError,
},
#[error("el motor de contenedores no contesta: {causa}")]
MotorDeContenedores {
#[source]
causa: crate::docker::Error,
},
#[error("el repositorio no ha respondido: {motivo}")]
Repositorio {
#[source]
causa: crate::rustic::Error,
motivo: String,
},
#[error("el trabajo contra el repositorio se ha interrumpido: {motivo}")]
TrabajoInterrumpido {
motivo: String,
},
#[error(
"el motor tiene que montarse desde dentro de un runtime de Tokio: es el runtime en el que \
sus hilos escriben en la base"
)]
SinRuntime,
}
#[derive(Clone)]
pub(crate) struct Contexto {
pub(crate) ajustes: Ajustes,
pub(crate) dependencias: Dependencias,
pub(crate) runtime: Handle,
pub(crate) hub: Hub,
}
impl Contexto {
pub(crate) fn en_la_base<T>(&self, futuro: impl std::future::Future<Output = T>) -> T {
self.runtime.block_on(futuro)
}
}
#[derive(Clone)]
pub struct Motor(Arc<Interior>);
struct Interior {
contexto: Contexto,
cola: Cola,
}
impl Motor {
pub fn arrancar(ajustes: Ajustes, dependencias: Dependencias) -> Result<Self, ErrorDelMotor> {
let runtime = Handle::try_current().map_err(|_| ErrorDelMotor::SinRuntime)?;
Ok(Self(Arc::new(Interior {
contexto: Contexto {
ajustes,
dependencias,
runtime,
hub: Hub::nuevo(),
},
cola: Cola::nueva(),
})))
}
pub async fn comprobar(&self) -> Result<(), ErrorDelMotor> {
let ajustes = &self.0.contexto.ajustes;
let cliente = Cliente::conectar(&ajustes.socket_del_motor, ajustes.espera_del_motor)
.map_err(|causa| ErrorDelMotor::MotorDeContenedores { causa })?;
cliente
.comprobar()
.await
.map_err(|causa| ErrorDelMotor::MotorDeContenedores { causa })
}
pub async fn encolar(&self, job_id: Uuid) -> Result<Ejecucion, ErrorDelMotor> {
let interior = &self.0;
let dependencias = &interior.contexto.dependencias;
if interior.cola.esta_parada() {
return Err(ErrorDelMotor::MotorParado);
}
let tarea = dependencias
.tareas
.leer(&job_id)
.await
.map_err(|causa| ErrorDelMotor::Base { causa })?
.ok_or(ErrorDelMotor::TareaQueNoEsta { job_id })?;
let ejecucion = dependencias
.ejecuciones
.alta_si_no_hay_ninguna_en_curso(AltaDeEjecucion {
job_id,
status: Estado::Pending,
snapshot_id: None,
bytes_processed: 0,
bytes_uploaded: 0,
error_message: None,
started_at: OffsetDateTime::now_utc(),
finished_at: None,
})
.await
.map_err(|causa| ErrorDelMotor::Base { causa })?
.ok_or(ErrorDelMotor::YaTieneUnaEnCurso { job_id })?;
if let Err(causa) =
interior
.cola
.encolar(&interior.contexto, tarea.repository_id, ejecucion.clone())
{
ejecucion::marcar_fallida_si_no_termino(
&interior.contexto,
&ejecucion.id,
&causa.to_string(),
)
.await;
return Err(causa);
}
interior.contexto.hub.publicar(EventoDeEjecucion {
execution_id: ejecucion.id,
job_id,
status: Estado::Pending,
bytes_processed: 0,
bytes_uploaded: 0,
percentage: None,
current_action: ejecucion::ACCION_ENCOLADA.to_owned(),
});
Ok(ejecucion)
}
pub fn hub(&self) -> &Hub {
&self.0.contexto.hub
}
pub fn cerrar_el_hub(&self) {
self.0.contexto.hub.cerrar();
}
pub async fn abrir_repositorio(&self, acceso: Acceso) -> Result<PuestaEnMarcha, ErrorDelMotor> {
let abridor = self.0.contexto.ajustes.abridor;
fuera_del_runtime(move || {
abridor(&acceso, BarrasDelMotor::mudas())
.map(|(_repositorio, puesta)| puesta)
.map_err(|causa| ErrorDelMotor::Repositorio {
motivo: ejecucion::motivo_de_la_biblioteca(&causa.to_string(), &acceso),
causa,
})
})
.await
}
pub async fn listar_snapshots(
&self,
acceso: Acceso,
ventana: usize,
) -> Result<Vec<ResumenDeSnapshot>, ErrorDelMotor> {
let abridor = self.0.contexto.ajustes.abridor;
fuera_del_runtime(move || {
let (repositorio, _puesta) =
abridor(&acceso, BarrasDelMotor::mudas()).map_err(|causa| {
ErrorDelMotor::Repositorio {
motivo: ejecucion::motivo_de_la_biblioteca(&causa.to_string(), &acceso),
causa,
}
})?;
repositorio
.listar_snapshots(ventana)
.map_err(|causa| ErrorDelMotor::Repositorio {
motivo: ejecucion::motivo_de_la_biblioteca(&causa.to_string(), &acceso),
causa,
})
})
.await
}
pub fn parar(&self, espera: Duration) {
self.0.cola.parar(espera);
}
}
impl std::fmt::Debug for Motor {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Motor")
.field("parado", &self.0.cola.esta_parada())
.field("capacidad_del_hub", &self.0.contexto.hub.capacidad())
.finish_non_exhaustive()
}
}
async fn fuera_del_runtime<T, F>(trabajo: F) -> Result<T, ErrorDelMotor>
where
T: Send + 'static,
F: FnOnce() -> Result<T, ErrorDelMotor> + Send + 'static,
{
match tokio::task::spawn_blocking(trabajo).await {
Ok(desenlace) => desenlace,
Err(caida) => {
tracing::error!(
%caida,
"el hilo que hablaba con el repositorio no ha llegado a término"
);
Err(ErrorDelMotor::TrabajoInterrumpido {
motivo: if caida.is_panic() {
"el hilo que hablaba con el repositorio se ha caído con un pánico".to_owned()
} else {
"el hilo que hablaba con el repositorio no ha llegado a término".to_owned()
},
})
}
}
}
#[cfg(test)]
mod tests {
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use rustic_core::repofile::SnapshotId;
use sqlx::SqlitePool;
use tokio::sync::broadcast::error::{RecvError, TryRecvError};
use uuid::Uuid;
use super::cola::CAPACIDAD_DE_LA_COLA;
use super::hub::EventoDeEjecucion;
use super::{Ajustes, Dependencias, ErrorDelMotor, Motor, ejecucion};
use crate::db::entities::{
AltaDeRepositorio, AltaDeTarea, AltaDeTarget, Ejecucion, Ejecuciones, Estado, Repositorios,
Tareas, Targets, TipoDeTarget,
};
use crate::rustic::{Acceso, BarrasDelMotor, Error, PuestaEnMarcha, Repositorio};
use crate::secret::SecretString;
use crate::test_support::{
CLAVE_SECRETA_DE_PRUEBA, DirectorioDePrueba, PASSPHRASE_DE_PRUEBA, abridor_local,
abrir_local, base_con_clave, destino_de_s3_que_rechaza, directorio_de_prueba,
sembrar_snapshots, snapshot_guardado, volcado_guardado,
};
const PLAZO_DE_ESPERA: Duration = Duration::from_secs(90);
const PLAZO_DEL_CONTENEDOR: Duration = Duration::from_secs(180);
const SOCKET_QUE_NO_EXISTE: &str = "unix:///no/existe/kesto-prueba.sock";
fn no_alcanzable(acceso: &Acceso) -> Error {
Error::DestinoNoAlcanzable {
destino: acceso.endpoint.clone(),
motivo: "el abridor de esta prueba no abre nada".to_owned(),
}
}
fn abridor_lento(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
std::thread::sleep(Duration::from_secs(1));
Err(no_alcanzable(acceso))
}
fn abridor_que_panica(
_acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
panic!(
"el abridor de esta prueba se cae a propósito, y el pánico no puede llegar a quien pregunta"
);
}
fn abridor_que_no_abre(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
Err(no_alcanzable(acceso))
}
static DENTRO: AtomicUsize = AtomicUsize::new(0);
static MAXIMO: AtomicUsize = AtomicUsize::new(0);
fn abridor_que_mide_la_serie(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
let dentro = DENTRO.fetch_add(1, Ordering::SeqCst) + 1;
MAXIMO.fetch_max(dentro, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(200));
DENTRO.fetch_sub(1, Ordering::SeqCst);
Err(no_alcanzable(acceso))
}
static ESPERANDO: AtomicUsize = AtomicUsize::new(0);
static MAXIMO_ESPERANDO: AtomicUsize = AtomicUsize::new(0);
static EN_PARALELO: AtomicBool = AtomicBool::new(false);
fn abridor_que_espera_al_otro(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
let dentro = ESPERANDO.fetch_add(1, Ordering::SeqCst) + 1;
MAXIMO_ESPERANDO.fetch_max(dentro, Ordering::SeqCst);
let limite = Instant::now() + Duration::from_secs(5);
while ESPERANDO.load(Ordering::SeqCst) < 2 && Instant::now() < limite {
std::thread::yield_now();
}
if ESPERANDO.load(Ordering::SeqCst) >= 2 {
EN_PARALELO.store(true, Ordering::SeqCst);
}
std::thread::sleep(Duration::from_millis(50));
ESPERANDO.fetch_sub(1, Ordering::SeqCst);
Err(no_alcanzable(acceso))
}
static LLAMADAS: AtomicUsize = AtomicUsize::new(0);
fn abridor_que_falla_la_primera(
acceso: &Acceso,
barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
if LLAMADAS.fetch_add(1, Ordering::SeqCst) == 0 {
return Err(no_alcanzable(acceso));
}
abridor_local(acceso, barras)
}
static TENIA_RUNTIME: AtomicBool = AtomicBool::new(true);
fn abridor_que_mira_el_runtime(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
TENIA_RUNTIME.store(
tokio::runtime::Handle::try_current().is_ok(),
Ordering::SeqCst,
);
Err(no_alcanzable(acceso))
}
fn abridor_que_duerme_tres_segundos(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
std::thread::sleep(Duration::from_secs(3));
Err(no_alcanzable(acceso))
}
fn abridor_que_no_termina(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
std::thread::sleep(Duration::from_secs(5));
Err(no_alcanzable(acceso))
}
fn abridor_con_total(
acceso: &Acceso,
barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
let barra = <BarrasDelMotor as rustic_core::ProgressBars>::progress(
&barras,
rustic_core::ProgressType::Bytes,
"prueba-del-motor",
);
barra.set_length(2000);
barra.inc(500);
Err(no_alcanzable(acceso))
}
fn abridor_sin_total(
acceso: &Acceso,
barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
let barra = <BarrasDelMotor as rustic_core::ProgressBars>::progress(
&barras,
rustic_core::ProgressType::Bytes,
"prueba-del-motor",
);
barra.inc(500);
Err(no_alcanzable(acceso))
}
static DESTINO_ESPERADO: OnceLock<String> = OnceLock::new();
static DESTINO_RECIBIDO: OnceLock<String> = OnceLock::new();
static EL_DESTINO_ERA_EL_ESPERADO: AtomicBool = AtomicBool::new(false);
static DENTRO_DEL_ABRIDOR: AtomicBool = AtomicBool::new(false);
static PUEDE_SEGUIR_EL_ABRIDOR: AtomicBool = AtomicBool::new(false);
fn abridor_que_apunta_el_destino_y_espera_la_senal(
acceso: &Acceso,
barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
let _ = DESTINO_RECIBIDO.set(acceso.endpoint.clone());
EL_DESTINO_ERA_EL_ESPERADO.store(
DESTINO_ESPERADO
.get()
.is_some_and(|esperado| *esperado == acceso.endpoint),
Ordering::SeqCst,
);
DENTRO_DEL_ABRIDOR.store(true, Ordering::SeqCst);
let limite = Instant::now() + Duration::from_secs(5);
while !PUEDE_SEGUIR_EL_ABRIDOR.load(Ordering::SeqCst) && Instant::now() < limite {
std::thread::yield_now();
}
abridor_local(acceso, barras)
}
static ATASCADO_EN_EL_ABRIDOR: AtomicBool = AtomicBool::new(false);
static PUEDE_SOLTARSE_EL_ABRIDOR: AtomicBool = AtomicBool::new(false);
static LLAMADAS_AL_ABRIDOR: AtomicUsize = AtomicUsize::new(0);
fn abridor_que_se_atasca_la_primera(
acceso: &Acceso,
_barras: BarrasDelMotor,
) -> Result<(Repositorio, PuestaEnMarcha), Error> {
if LLAMADAS_AL_ABRIDOR.fetch_add(1, Ordering::SeqCst) == 0 {
ATASCADO_EN_EL_ABRIDOR.store(true, Ordering::SeqCst);
let limite = Instant::now() + Duration::from_secs(5);
while !PUEDE_SOLTARSE_EL_ABRIDOR.load(Ordering::SeqCst) && Instant::now() < limite {
std::thread::yield_now();
}
}
Err(no_alcanzable(acceso))
}
struct Aparato {
_base: DirectorioDePrueba,
pool: SqlitePool,
dependencias: Dependencias,
_destino: DirectorioDePrueba,
destino: PathBuf,
fuentes: Vec<DirectorioDePrueba>,
otros_destinos: Vec<DirectorioDePrueba>,
repositorio: Uuid,
}
impl Aparato {
async fn nuevo() -> Self {
let (base, pool, clave) = base_con_clave().await;
let destino_guardado = directorio_de_prueba();
let destino = destino_guardado.ruta().to_path_buf();
let dependencias = Dependencias {
repositorios: Arc::new(Repositorios::nuevo(pool.clone(), clave.clone())),
targets: Arc::new(Targets::nuevo(pool.clone(), clave.clone())),
tareas: Arc::new(Tareas::nuevo(pool.clone())),
ejecuciones: Arc::new(Ejecuciones::nuevo(pool.clone())),
};
let mut aparato = Self {
_base: base,
pool,
dependencias,
_destino: destino_guardado,
destino,
fuentes: Vec::new(),
otros_destinos: Vec::new(),
repositorio: Uuid::nil(),
};
aparato.repositorio = aparato
.dar_de_alta_el_repositorio("repositorio-de-prueba", &aparato.destino.clone())
.await;
aparato
}
async fn dar_de_alta_el_repositorio_con_endpoint(
&self,
nombre: &str,
endpoint: &str,
) -> Uuid {
self.dependencias
.repositorios
.alta(AltaDeRepositorio {
name: nombre.to_owned(),
endpoint: endpoint.to_owned(),
bucket: "cubo-de-prueba".to_owned(),
region: "region-de-prueba".to_owned(),
access_key_id: "clave-de-acceso-de-prueba".to_owned(),
secret_access_key: SecretString::new(CLAVE_SECRETA_DE_PRUEBA.to_owned()),
passphrase: SecretString::new(PASSPHRASE_DE_PRUEBA.to_owned()),
})
.await
.expect("el alta del repositorio no puede fallar")
.id
}
async fn dar_de_alta_el_repositorio(&self, nombre: &str, endpoint: &Path) -> Uuid {
self.dar_de_alta_el_repositorio_con_endpoint(nombre, &endpoint.to_string_lossy())
.await
}
async fn segundo_destino(&mut self) -> Uuid {
let guardado = directorio_de_prueba();
let ruta = guardado.ruta().to_path_buf();
let id = self
.dar_de_alta_el_repositorio_con_endpoint(
"repositorio-de-al-lado",
&ruta.to_string_lossy(),
)
.await;
self.otros_destinos.push(guardado);
id
}
async fn acceso_de(&self, repository_id: &Uuid) -> Acceso {
let fila = self
.dependencias
.repositorios
.leer(repository_id)
.await
.expect("la lectura del repositorio no puede fallar")
.expect("el repositorio de la prueba tiene que estar");
Acceso {
endpoint: fila.endpoint,
bucket: fila.bucket,
region: fila.region,
access_key_id: fila.access_key_id,
secret_access_key: fila.secret_access_key,
passphrase: fila.passphrase,
}
}
async fn apuntar_la_tarea_a(&self, tarea: &Uuid, repositorio: Uuid) {
let mut fila = self
.dependencias
.tareas
.leer(tarea)
.await
.expect("la lectura de la tarea no puede fallar")
.expect("la tarea de la prueba tiene que estar");
fila.repository_id = repositorio;
let editada = self
.dependencias
.tareas
.actualizar(&fila)
.await
.expect("la edición de la tarea no puede fallar");
assert!(
editada,
"la tarea de la prueba tiene que haber quedado editada, o la ventana que se mide \
no se habría abierto"
);
}
async fn segundo_repositorio(&self) -> Uuid {
let endpoint = PathBuf::from(format!("/no/existe/kesto-prueba-{}", Uuid::new_v4()));
self.dar_de_alta_el_repositorio("repositorio-hermano", &endpoint)
.await
}
async fn target_de_directorio(&mut self, ficheros: &[(&str, &str)]) -> Uuid {
let fuente = directorio_de_prueba();
for (nombre, contenido) in ficheros {
std::fs::write(fuente.ruta().join(nombre), contenido)
.expect("el fichero de la fuente se escribe");
}
let id = self
.dependencias
.targets
.alta(AltaDeTarget {
name: "target-de-prueba".to_owned(),
target_type: TipoDeTarget::Directory,
container_name: None,
db_user: None,
db_password: None,
db_name: None,
host_path: Some(fuente.ruta().to_string_lossy().into_owned()),
})
.await
.expect("el alta del target no puede fallar")
.id;
self.fuentes.push(fuente);
id
}
fn la_fuente(&self) -> &Path {
self.fuentes
.last()
.expect("esta prueba tiene que haber dado de alta una fuente")
.ruta()
}
async fn target_de_base_sin_contenedor(&self) -> Uuid {
self.dependencias
.targets
.alta(AltaDeTarget {
name: "target-incoherente".to_owned(),
target_type: TipoDeTarget::Mariadb,
container_name: None,
db_user: Some("operador".to_owned()),
db_password: None,
db_name: None,
host_path: None,
})
.await
.expect("el alta del target no puede fallar")
.id
}
async fn target_de_contenedor(&self, contenedor: &str, contrasena: Option<&str>) -> Uuid {
self.dependencias
.targets
.alta(AltaDeTarget {
name: "tienda".to_owned(),
target_type: TipoDeTarget::Mariadb,
container_name: Some(contenedor.to_owned()),
db_user: Some("operador".to_owned()),
db_password: contrasena.map(|valor| SecretString::new(valor.to_owned())),
db_name: None,
host_path: None,
})
.await
.expect("el alta del target no puede fallar")
.id
}
async fn tarea(
&self,
target_id: Uuid,
repository_id: Uuid,
retencion: (i64, i64, i64),
enabled: bool,
) -> Uuid {
let (keep_daily, keep_weekly, keep_monthly) = retencion;
self.dependencias
.tareas
.alta(AltaDeTarea {
target_id,
repository_id,
cron_expression: "0 3 * * *".to_owned(),
keep_daily,
keep_weekly,
keep_monthly,
enabled,
})
.await
.expect("el alta de la tarea no puede fallar")
.id
}
async fn tarea_normal(&self, target_id: Uuid) -> Uuid {
let repositorio = self.repositorio;
self.tarea(target_id, repositorio, (7, 4, 12), true).await
}
fn motor_con(&self, abridor: super::AbridorDeRepositorios) -> Motor {
Motor::arrancar(
Ajustes::nuevos(SOCKET_QUE_NO_EXISTE).con_abridor(abridor),
self.dependencias.clone(),
)
.expect("el motor se monta desde dentro del runtime de la prueba")
}
fn motor_y_ajustes(&self, ajustes: Ajustes) -> Motor {
Motor::arrancar(ajustes, self.dependencias.clone())
.expect("el motor se monta desde dentro del runtime de la prueba")
}
async fn esperar_el_final(&self, id: &Uuid) -> Ejecucion {
esperar_el_final(&self.dependencias.ejecuciones, id).await
}
async fn filas_en(&self, estado: Estado) -> i64 {
sqlx::query_scalar("SELECT COUNT(*) FROM job_executions WHERE status = ?1")
.bind(estado.texto())
.fetch_one(&self.pool)
.await
.expect("el recuento de filas debe poder leerse")
}
async fn estado_en_la_base(&self, id: &Uuid) -> String {
sqlx::query_scalar("SELECT status FROM job_executions WHERE id = ?1")
.bind(id.hyphenated().to_string())
.fetch_one(&self.pool)
.await
.expect("el estado de la fila debe poder leerse")
}
async fn fila(&self, id: &Uuid) -> Ejecucion {
self.dependencias
.ejecuciones
.leer(id)
.await
.expect("la lectura no puede fallar")
.expect("la fila tiene que seguir ahí")
}
fn instantaneas(&self) -> usize {
abrir_local(&self.destino, PASSPHRASE_DE_PRUEBA)
.list::<SnapshotId>()
.expect("el repositorio local de prueba lista sus instantáneas")
.count()
}
}
async fn esperar_el_final(ejecuciones: &Ejecuciones, id: &Uuid) -> Ejecucion {
esperar_el_final_con(ejecuciones, id, PLAZO_DE_ESPERA).await
}
async fn esperar_el_final_con(
ejecuciones: &Ejecuciones,
id: &Uuid,
plazo: Duration,
) -> Ejecucion {
let limite = Instant::now() + plazo;
loop {
let fila = ejecuciones
.leer(id)
.await
.expect("la lectura no puede fallar")
.expect("la fila tiene que seguir ahí");
if matches!(fila.status, Estado::Success | Estado::Failed) {
return fila;
}
assert!(
Instant::now() < limite,
"la ejecución no ha llegado a un estado terminal en {plazo:?}: {:?}",
fila.status
);
tokio::time::sleep(Duration::from_millis(10)).await;
}
}
async fn esperar_el_estado(aparato: &Aparato, id: &Uuid, estado: Estado) {
let limite = Instant::now() + PLAZO_DE_ESPERA;
loop {
if aparato.fila(id).await.status == estado {
return;
}
assert!(
Instant::now() < limite,
"la ejecución no llegó a `{}` en {PLAZO_DE_ESPERA:?}",
estado.texto()
);
tokio::time::sleep(Duration::from_millis(5)).await;
}
}
async fn esperar_a(señal: &AtomicBool, que: &str) {
let limite = Instant::now() + PLAZO_DE_ESPERA;
while !señal.load(Ordering::SeqCst) {
assert!(
Instant::now() < limite,
"no ha ocurrido {que} dentro de {PLAZO_DE_ESPERA:?}"
);
tokio::time::sleep(Duration::from_millis(5)).await;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn encolar_crea_la_fila_y_la_encola() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let ocupada = aparato.tarea_normal(target).await;
let en_cola = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_lento);
let primera = motor
.encolar(ocupada)
.await
.expect("la primera tiene que entrar");
assert_eq!(
primera.status,
Estado::Pending,
"una ejecución recién encolada nace en `pending`"
);
esperar_el_estado(&aparato, &primera.id, Estado::Running).await;
let segunda = motor
.encolar(en_cola)
.await
.expect("la segunda tarea tiene que entrar");
assert_eq!(
segunda.status,
Estado::Pending,
"la ejecución devuelta por el encolado no está en `pending`"
);
assert_eq!(
aparato.estado_en_la_base(&segunda.id).await,
"pending",
"la fila de la ejecución en cola no quedó en `pending` en la tabla"
);
assert_eq!(
aparato.filas_en(Estado::Pending).await,
1,
"tiene que haber exactamente una ejecución esperando en la cola"
);
assert_eq!(
aparato.esperar_el_final(&segunda.id).await.status,
Estado::Failed
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn encolar_sobre_una_tarea_que_no_existe_da_el_error_y_no_crea_fila() {
let aparato = Aparato::nuevo().await;
let motor = aparato.motor_con(abridor_local);
let inexistente = Uuid::new_v4();
let error = motor
.encolar(inexistente)
.await
.expect_err("una tarea que no existe no puede encolarse");
assert!(
matches!(error, ErrorDelMotor::TareaQueNoEsta { job_id } if job_id == inexistente),
"el error no es el de «la tarea no está»: {error:?}"
);
assert!(
error.to_string().contains(&inexistente.to_string()),
"el motivo no nombra la tarea que falta: {error}"
);
let filas: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM job_executions")
.fetch_one(&aparato.pool)
.await
.expect("el recuento debe poder leerse");
assert_eq!(filas, 0, "la tarea no existe y se creó una ejecución");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn encolar_dos_veces_sobre_la_misma_tarea_da_el_error_y_no_crea_una_segunda_fila() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let tarea = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_lento);
let primera = motor
.encolar(tarea)
.await
.expect("la primera tiene que entrar");
esperar_el_estado(&aparato, &primera.id, Estado::Running).await;
let error = motor
.encolar(tarea)
.await
.expect_err("con una ejecución en curso, la segunda no puede entrar");
assert!(
matches!(error, ErrorDelMotor::YaTieneUnaEnCurso { job_id } if job_id == tarea),
"el error no es el del `409`: {error:?}"
);
assert_eq!(
aparato.filas_en(Estado::Running).await + aparato.filas_en(Estado::Pending).await,
1,
"la guarda dijo que no, pero dejó una segunda fila en la tabla"
);
let filas: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM job_executions")
.fetch_one(&aparato.pool)
.await
.expect("el recuento debe poder leerse");
assert_eq!(
filas, 1,
"se creó una segunda ejecución para la misma tarea"
);
assert_eq!(
aparato.esperar_el_final(&primera.id).await.status,
Estado::Failed
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn una_tarea_desactivada_se_encola_igual() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let repositorio = aparato.repositorio;
let desactivada = aparato.tarea(target, repositorio, (7, 4, 12), false).await;
let motor = aparato.motor_con(abridor_lento);
let ejecucion = motor
.encolar(desactivada)
.await
.expect("una tarea desactivada sí se dispara a mano");
assert_eq!(ejecucion.status, Estado::Pending);
assert_eq!(
aparato.esperar_el_final(&ejecucion.id).await.status,
Estado::Failed
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_ciclo_de_vida_completo_copia_de_verdad() {
let mut aparato = Aparato::nuevo().await;
let target = aparato
.target_de_directorio(&[("uno.txt", "contenido uno"), ("dos.txt", "dos")])
.await;
let fuente = aparato.la_fuente().to_path_buf();
let tarea = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_local);
let ejecucion = motor
.encolar(tarea)
.await
.expect("el encolado tiene que entrar");
assert_eq!(
ejecucion.status,
Estado::Pending,
"la ejecución no nace en `pending`"
);
let en_la_tabla = aparato.estado_en_la_base(&ejecucion.id).await;
assert!(
en_la_tabla == Estado::Pending.texto() || en_la_tabla == Estado::Running.texto(),
"la ejecución no está en curso en la tabla, sino en `{en_la_tabla}`"
);
let final_de_la_fila = aparato.esperar_el_final(&ejecucion.id).await;
assert_eq!(
final_de_la_fila.status,
Estado::Success,
"la copia tenía que acabar bien y acabó en {:?}: {:?}",
final_de_la_fila.status,
final_de_la_fila.error_message
);
assert!(
final_de_la_fila.finished_at.is_some(),
"una ejecución terminada tiene que llevar su `finished_at`"
);
let instantanea = final_de_la_fila
.snapshot_id
.clone()
.expect("una copia buena declara la instantánea que ha guardado");
assert!(
final_de_la_fila.bytes_processed > 0 && final_de_la_fila.bytes_uploaded > 0,
"los contadores de una copia de directorio no pueden salir a cero: {}, {}",
final_de_la_fila.bytes_processed,
final_de_la_fila.bytes_uploaded
);
let nodo = fuente
.join("uno.txt")
.to_string_lossy()
.trim_start_matches('/')
.to_owned();
let guardado =
volcado_guardado(&aparato.destino, PASSPHRASE_DE_PRUEBA, &instantanea, &nodo);
assert_eq!(
guardado.as_deref(),
Some(b"contenido uno".as_slice()),
"el fichero copiado no está en el repositorio, o no tiene el contenido que se copió"
);
let resumen = snapshot_guardado(&aparato.destino, PASSPHRASE_DE_PRUEBA, &instantanea);
assert!(
resumen.summary.is_some(),
"la instantánea guardada no trae resumen: el archivado no procesó nada"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn un_fallo_acaba_en_failed_con_su_motivo_y_ninguna_ejecucion_se_queda_en_running() {
let mut aparato = Aparato::nuevo().await;
let incoherente = aparato.target_de_base_sin_contenedor().await;
let con_destino_roto = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let sin_contenedor = aparato.tarea_normal(incoherente).await;
let sin_destino = aparato.tarea_normal(con_destino_roto).await;
let motor = aparato.motor_con(abridor_que_no_abre);
let primera = motor
.encolar(sin_contenedor)
.await
.expect("tiene que entrar");
let segunda = motor.encolar(sin_destino).await.expect("tiene que entrar");
let primera = aparato.esperar_el_final(&primera.id).await;
let segunda = aparato.esperar_el_final(&segunda.id).await;
for fila in [&primera, &segunda] {
assert_eq!(fila.status, Estado::Failed, "el fallo no acabó en `failed`");
assert!(
fila.finished_at.is_some(),
"una ejecución fallida tiene que llevar su `finished_at`"
);
assert!(
fila.error_message.as_deref().is_some_and(|m| !m.is_empty()),
"una ejecución fallida tiene que decir por qué"
);
}
let motivo_incoherente = primera.error_message.clone().unwrap_or_default();
let motivo_del_destino = segunda.error_message.clone().unwrap_or_default();
assert!(
motivo_incoherente.contains("container_name"),
"el motivo de un objetivo incoherente tiene que nombrar el campo que falta: \
{motivo_incoherente}"
);
assert!(
motivo_del_destino.contains(&aparato.destino.to_string_lossy().to_string())
|| motivo_del_destino.contains("no se alcanza")
|| motivo_del_destino.contains("destino"),
"el motivo del destino roto no dice de qué va: {motivo_del_destino}"
);
assert_ne!(
motivo_incoherente, motivo_del_destino,
"los dos fallos tienen motivos distintos y la fila los guarda iguales"
);
assert_eq!(
aparato.filas_en(Estado::Running).await,
0,
"alguna ejecución se ha quedado en `running`: el panel la pintaría como «en marcha»"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn la_retencion_se_aplica_en_el_camino_bueno() {
let mut aparato = Aparato::nuevo().await;
sembrar_snapshots(
&aparato.destino,
PASSPHRASE_DE_PRUEBA,
&[
"2026-01-01T10:00:00Z",
"2026-01-02T10:00:00Z",
"2026-01-03T10:00:00Z",
],
);
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let repositorio = aparato.repositorio;
let tarea = aparato.tarea(target, repositorio, (1, 0, 0), true).await;
let motor = aparato.motor_con(abridor_local);
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
let final_de_la_fila = aparato.esperar_el_final(&ejecucion.id).await;
assert_eq!(
final_de_la_fila.status,
Estado::Success,
"la copia tenía que acabar bien: {:?}",
final_de_la_fila.error_message
);
let instantaneas = aparato.instantaneas();
assert_eq!(
instantaneas, 2,
"el repositorio quedó con {instantaneas} instantáneas y no con dos: la regla diaria \
tenía que retirar dos de las tres sembradas, y la copia nueva quedarse"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn un_fallo_de_la_retencion_no_cambia_el_resultado_de_la_copia() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let repositorio = aparato.repositorio;
let tarea = aparato.tarea(target, repositorio, (0, 0, 0), true).await;
let motor = aparato.motor_con(abridor_local);
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
let final_de_la_fila = aparato.esperar_el_final(&ejecucion.id).await;
assert_eq!(
final_de_la_fila.status,
Estado::Success,
"la copia estaba hecha y la retención falló: la fila no puede decir `failed`: {:?}",
final_de_la_fila.error_message
);
assert!(
final_de_la_fila.snapshot_id.is_some(),
"una copia buena declara su instantánea aunque la retención falle"
);
assert_eq!(
final_de_la_fila.error_message, None,
"una copia buena no puede llevar motivo de error porque falle la retención"
);
assert_eq!(
aparato.instantaneas(),
1,
"la retención falló y aun así retiró algo"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dos_ejecuciones_del_mismo_repositorio_van_en_serie() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let una = aparato.tarea_normal(target).await;
let otra = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_que_mide_la_serie);
let primera = motor
.encolar(una)
.await
.expect("la primera tiene que entrar");
let segunda = motor
.encolar(otra)
.await
.expect("la segunda tiene que entrar");
assert_eq!(
aparato.esperar_el_final(&primera.id).await.status,
Estado::Failed
);
assert_eq!(
aparato.esperar_el_final(&segunda.id).await.status,
Estado::Failed
);
assert_eq!(
MAXIMO.load(Ordering::SeqCst),
1,
"dos ejecuciones del mismo repositorio estuvieron dentro del abridor a la vez: el \
trabajador tiene que atenderlas de una en una"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dos_ejecuciones_de_repositorios_distintos_pueden_estar_a_la_vez_en_running() {
let mut aparato = Aparato::nuevo().await;
let hermana = aparato.segundo_repositorio().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let repositorio = aparato.repositorio;
let una = aparato.tarea(target, repositorio, (7, 4, 12), true).await;
let otra = aparato.tarea(target, hermana, (7, 4, 12), true).await;
let motor = aparato.motor_con(abridor_que_espera_al_otro);
let primera = motor
.encolar(una)
.await
.expect("la primera tiene que entrar");
let segunda = motor
.encolar(otra)
.await
.expect("la segunda tiene que entrar");
assert_eq!(
aparato.esperar_el_final(&primera.id).await.status,
Estado::Failed
);
assert_eq!(
aparato.esperar_el_final(&segunda.id).await.status,
Estado::Failed
);
assert!(
EN_PARALELO.load(Ordering::SeqCst),
"los dos repositorios no llegaron a estar a la vez dentro del abridor: dos \
repositorios distintos no tienen por qué esperarse"
);
assert_eq!(
MAXIMO_ESPERANDO.load(Ordering::SeqCst),
2,
"el máximo de ejecuciones simultáneas no fue dos"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_orden_de_encolado_se_respeta() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let motor = aparato.motor_con(abridor_lento);
let mut encoladas = Vec::new();
for _ in 0..3 {
let tarea = aparato.tarea_normal(target).await;
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
encoladas.push(ejecucion.id);
}
let limite = Instant::now() + PLAZO_DE_ESPERA;
let mut observadas: Vec<Uuid> = Vec::new();
while observadas.len() < encoladas.len() && Instant::now() < limite {
let filas = aparato
.dependencias
.ejecuciones
.listar()
.await
.expect("el listado no puede fallar");
for fila in filas {
if fila.status == Estado::Running && !observadas.contains(&fila.id) {
observadas.push(fila.id);
}
}
tokio::time::sleep(Duration::from_millis(5)).await;
}
assert_eq!(
observadas, encoladas,
"las tres ejecuciones no se atendieron en el orden en que se encolaron"
);
for id in &encoladas {
assert_eq!(aparato.esperar_el_final(id).await.status, Estado::Failed);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn un_fallo_no_detiene_al_trabajador() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let falla = aparato.tarea_normal(target).await;
let va_bien = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_que_falla_la_primera);
let primera = motor.encolar(falla).await.expect("tiene que entrar");
assert_eq!(
aparato.esperar_el_final(&primera.id).await.status,
Estado::Failed,
"la primera tenía que fallar"
);
let segunda = motor.encolar(va_bien).await.expect("tiene que entrar");
let final_de_la_fila = aparato.esperar_el_final(&segunda.id).await;
assert_eq!(
final_de_la_fila.status,
Estado::Success,
"después de un fallo, la siguiente ejecución del mismo repositorio tiene que arrancar \
y acabar bien: {:?}",
final_de_la_fila.error_message
);
assert!(
final_de_la_fila.snapshot_id.is_some(),
"la copia que fue bien no dejó su instantánea"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_trabajador_no_tiene_runtime_cuando_abre_el_repositorio() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let tarea = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_que_mira_el_runtime);
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
assert_eq!(
aparato.esperar_el_final(&ejecucion.id).await.status,
Estado::Failed
);
assert!(
!TENIA_RUNTIME.load(Ordering::SeqCst),
"el hilo del trabajador estaba dentro de un runtime al abrir el repositorio: llamar a \
Rustic desde ahí entra en pánico, y el pánico deja la copia sin hacer"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_trabajo_no_ocupa_un_hilo_de_tokio() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let tarea = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_que_duerme_tres_segundos);
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
esperar_el_estado(&aparato, &ejecucion.id, Estado::Running).await;
tokio::time::sleep(Duration::from_millis(100)).await;
let consulta = tokio::time::timeout(
Duration::from_secs(1),
aparato.dependencias.ejecuciones.leer(&ejecucion.id),
)
.await;
assert!(
consulta.is_ok(),
"una consulta a la base desde el runtime no se atendió en un segundo: el trabajo de la \
copia está ocupando un hilo de Tokio"
);
assert!(
matches!(consulta.expect("ya se comprobó"), Ok(Some(_))),
"la consulta no devolvió la fila de la ejecución"
);
assert_eq!(
aparato.fila(&ejecucion.id).await.status,
Estado::Running,
"la copia ya había terminado cuando se hizo la consulta: la medición no dice nada"
);
assert_eq!(
aparato.esperar_el_final(&ejecucion.id).await.status,
Estado::Failed
);
}
fn recoger(
receptor: &mut tokio::sync::broadcast::Receiver<EventoDeEjecucion>,
) -> Vec<EventoDeEjecucion> {
let mut eventos = Vec::new();
loop {
match receptor.try_recv() {
Ok(evento) => eventos.push(evento),
Err(TryRecvError::Empty | TryRecvError::Closed) => break,
Err(TryRecvError::Lagged(_)) => continue,
}
}
eventos
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_porcentaje_es_nulo_sin_total_y_numerico_con_total() {
let mut aparato = Aparato::nuevo().await;
let con_total = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let sin_total = aparato.target_de_directorio(&[("dos.txt", "dos")]).await;
let tarea_con = aparato.tarea_normal(con_total).await;
let tarea_sin = aparato.tarea_normal(sin_total).await;
let motor = aparato.motor_con(abridor_con_total);
let mut receptor = motor
.hub()
.suscribir()
.expect("el hub no está cerrado en esta prueba");
let ejecucion = motor.encolar(tarea_con).await.expect("tiene que entrar");
let fila = aparato.esperar_el_final(&ejecucion.id).await;
let eventos = recoger(&mut receptor);
assert!(
eventos.len() >= 2,
"se esperaban al menos el de `pending` y el final"
);
assert_eq!(
eventos[0].status,
Estado::Pending,
"el primer evento no es el nacimiento de la ejecución"
);
assert_eq!(eventos[0].current_action, ejecucion::ACCION_ENCOLADA);
let final_del_evento = eventos.last().expect("hay eventos");
assert_eq!(
final_del_evento.execution_id, ejecucion.id,
"el evento final no es de la ejecución que corrió"
);
assert_eq!(
final_del_evento.job_id, tarea_con,
"el evento no lleva la tarea de la ejecución"
);
assert_eq!(
final_del_evento.status, fila.status,
"el `status` del evento no coincide con el de la fila"
);
assert_eq!(fila.status, Estado::Failed);
assert_eq!(
final_del_evento.bytes_processed, 2000,
"el total que declaró el gancho no llegó al evento como bytes procesados"
);
assert_eq!(
final_del_evento.bytes_uploaded, 500,
"los bytes que informó el gancho no llegaron al evento"
);
assert_eq!(
final_del_evento.percentage,
Some(25.0),
"con un total declarado y mayor que cero, el porcentaje tiene que ser un número"
);
assert!(
!final_del_evento.current_action.is_empty(),
"el evento no dice qué estaba haciendo la ejecución"
);
assert_eq!(
fila.bytes_processed, 2000,
"los contadores del evento y los de la fila tienen que ser los mismos"
);
assert_eq!(fila.bytes_uploaded, 500);
let motor = aparato.motor_con(abridor_sin_total);
let mut receptor = motor
.hub()
.suscribir()
.expect("el hub no está cerrado en esta prueba");
let ejecucion = motor.encolar(tarea_sin).await.expect("tiene que entrar");
let fila = aparato.esperar_el_final(&ejecucion.id).await;
let eventos = recoger(&mut receptor);
let final_del_evento = eventos.last().expect("hay eventos");
assert_eq!(
final_del_evento.percentage, None,
"sin un total declarado el porcentaje tiene que ser nulo: un porcentaje sin \
denominador es una cifra inventada"
);
assert_eq!(
final_del_evento.bytes_processed, 500,
"sin total, los bytes procesados son los que se han leído, y siguen informados"
);
assert_eq!(final_del_evento.bytes_uploaded, 500);
assert_eq!(fila.status, final_del_evento.status);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parar_deja_de_aceptar_trabajo_y_espera_lo_acotado() {
const ESPERA: Duration = Duration::from_millis(200);
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let tarea = aparato.tarea_normal(target).await;
let mut ajustes = Ajustes::nuevos(SOCKET_QUE_NO_EXISTE).con_abridor(abridor_que_no_termina);
ajustes.espera_de_parada = ESPERA;
let motor = aparato.motor_y_ajustes(ajustes);
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
esperar_el_estado(&aparato, &ejecucion.id, Estado::Running).await;
let (buffer, guardia) = crate::test_support::capturar_log();
let empezado = Instant::now();
motor.parar(ESPERA);
let tardado = empezado.elapsed();
let registro = buffer.texto();
drop(guardia);
assert!(
tardado < Duration::from_secs(2),
"la parada no respetó su plazo: tardó {tardado:?}"
);
assert!(
tardado >= ESPERA,
"la parada volvió antes de que se le acabara el plazo: {tardado:?}"
);
assert!(
registro.contains("plazo de parada"),
"la parada no dejó constancia de la copia que no terminó: {registro}"
);
let error = motor
.encolar(tarea)
.await
.expect_err("un motor parado no puede aceptar trabajo");
assert!(
matches!(error, ErrorDelMotor::MotorParado),
"el error de un motor parado no es el que dice que está parado: {error:?}"
);
tokio::time::sleep(Duration::from_millis(1200)).await;
assert_eq!(
aparato.esperar_el_final(&ejecucion.id).await.status,
Estado::Failed,
"la copia abandonada por la parada no llegó a escribir su final"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cerrar_el_hub_termina_el_stream_aunque_haya_una_ejecucion_en_vuelo() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let tarea = aparato.tarea_normal(target).await;
let motor = aparato.motor_con(abridor_que_no_termina);
let mut cliente = motor
.hub()
.suscribir()
.expect("el hub no está cerrado antes de la ejecución");
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
esperar_el_estado(&aparato, &ejecucion.id, Estado::Running).await;
motor.cerrar_el_hub();
assert_eq!(
aparato.estado_en_la_base(&ejecucion.id).await,
Estado::Running.texto(),
"la copia terminó antes del cierre: esta prueba solo mide el caso del estado en vivo si \
la ejecución sigue en vuelo cuando se cierra el hub"
);
let limite = Instant::now() + Duration::from_secs(5);
let desenlace = loop {
let restante = limite.saturating_duration_since(Instant::now());
match tokio::time::timeout(restante, cliente.recv()).await {
Ok(Ok(_)) => continue,
Ok(Err(error)) => break error,
Err(caducado) => panic!(
"el cierre del hub no despertó al cliente con una ejecución en vuelo dentro del \
plazo: el clon del hub que guarda su estado en vivo mantiene el canal abierto, \
el cuerpo del stream no termina y el apagado ordenado se queda esperándolo \
({caducado})"
),
}
};
assert!(
matches!(desenlace, RecvError::Closed),
"el cliente del stream no vio su canal cerrado, sino {desenlace:?}"
);
assert!(
motor.hub().suscribir().is_none(),
"un hub cerrado siguió dando receptores"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn una_ejecucion_que_no_puede_empezar_acaba_en_failed_y_no_bloquea_su_tarea() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let tarea = aparato.tarea_normal(target).await;
sqlx::query(
"CREATE TRIGGER bloquea_el_running BEFORE UPDATE ON job_executions \
FOR EACH ROW WHEN NEW.status = 'running' \
BEGIN SELECT RAISE(ABORT, 'el paso a `running` está bloqueado por la prueba'); END",
)
.execute(&aparato.pool)
.await
.expect("el disparador de la prueba se crea");
let motor = aparato.motor_con(abridor_local);
let ejecucion = motor
.encolar(tarea)
.await
.expect("el alta tiene que entrar");
let fila = aparato.esperar_el_final(&ejecucion.id).await;
assert_eq!(
fila.status,
Estado::Failed,
"la fila se quedó en {:?} con `finished_at` {:?}: una ejecución que no puede empezar \
tiene que acabar en `failed`, o su tarea no se puede volver a disparar y el panel \
miente diciendo que está en cola",
fila.status,
fila.finished_at
);
assert!(
fila.finished_at.is_some(),
"una ejecución que acaba tiene que llevar su `finished_at`, y ésta no lo llevaba"
);
let motivo = fila.error_message.clone().unwrap_or_default();
assert!(
motivo.contains("`running`"),
"el motivo tiene que decir que no se pudo pasar a `running`: {motivo}"
);
let segunda = motor.encolar(tarea).await.expect(
"la tarea tiene que poder volver a dispararse después de una ejecución que no empezó",
);
assert_eq!(segunda.status, Estado::Pending);
assert_eq!(
aparato.esperar_el_final(&segunda.id).await.status,
Estado::Failed
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn la_cola_llena_da_un_error_con_nombre_y_su_fila_no_se_queda_en_pending() {
let mut aparato = Aparato::nuevo().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let motor = aparato.motor_con(abridor_que_se_atasca_la_primera);
let primera = aparato.tarea_normal(target).await;
let mut encoladas = vec![motor.encolar(primera).await.expect("tiene que entrar")];
esperar_a(
&ATASCADO_EN_EL_ABRIDOR,
"que el trabajador se atasque dentro del abridor",
)
.await;
for _ in 0..CAPACIDAD_DE_LA_COLA {
let tarea = aparato.tarea_normal(target).await;
encoladas.push(
motor
.encolar(tarea)
.await
.expect("tiene que caber en la cola"),
);
}
assert_eq!(
aparato.filas_en(Estado::Pending).await,
CAPACIDAD_DE_LA_COLA as i64,
"la cola no quedó llena de verdad, así que el encolado de abajo no mediría lo que dice"
);
let de_mas = aparato.tarea_normal(target).await;
let error = motor
.encolar(de_mas)
.await
.expect_err("con la cola llena, la siguiente no puede entrar");
assert!(
matches!(
error,
ErrorDelMotor::ColaDelRepositorioLlena { repository_id }
if repository_id == aparato.repositorio
),
"el error de una cola llena no es el que la nombra: {error:?}"
);
let (estado, finished_at): (String, Option<String>) =
sqlx::query_as("SELECT status, finished_at FROM job_executions WHERE job_id = ?1")
.bind(de_mas.hyphenated().to_string())
.fetch_one(&aparato.pool)
.await
.expect("la fila de la ejecución que no cupo tiene que existir");
assert_eq!(
estado, "failed",
"la fila de la ejecución que no cupo se quedó en `{estado}`: el panel la pintaría \
esperando una copia que nadie va a hacer"
);
assert!(
finished_at.is_some(),
"la fila de la ejecución que no cupo no llevaba `finished_at`"
);
PUEDE_SOLTARSE_EL_ABRIDOR.store(true, Ordering::SeqCst);
let ultima = encoladas.last().expect("la prueba encoló al menos una");
assert_eq!(
aparato.esperar_el_final(&ultima.id).await.status,
Estado::Failed
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn la_copia_va_al_repositorio_del_encolado_aunque_la_tarea_se_edite_entre_medias() {
let mut aparato = Aparato::nuevo().await;
let al_lado = aparato.segundo_destino().await;
let target = aparato.target_de_directorio(&[("uno.txt", "uno")]).await;
let repositorio = aparato.repositorio;
let tarea = aparato.tarea(target, repositorio, (7, 4, 12), true).await;
let motor = aparato.motor_con(abridor_que_apunta_el_destino_y_espera_la_senal);
let _ = DESTINO_ESPERADO.set(aparato.destino.to_string_lossy().into_owned());
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
esperar_a(&DENTRO_DEL_ABRIDOR, "que el trabajador entre en el abridor").await;
aparato.apuntar_la_tarea_a(&tarea, al_lado).await;
PUEDE_SEGUIR_EL_ABRIDOR.store(true, Ordering::SeqCst);
let fila = aparato.esperar_el_final(&ejecucion.id).await;
assert_eq!(
fila.status,
Estado::Success,
"la copia tenía que acabar bien y acabó en {:?}: {:?}",
fila.status,
fila.error_message
);
assert!(
EL_DESTINO_ERA_EL_ESPERADO.load(Ordering::SeqCst),
"el abridor recibió el destino {:?} y el que decidió la serialización era {:?}: la \
copia fue a un repositorio que nadie serializó",
DESTINO_RECIBIDO.get(),
DESTINO_ESPERADO.get()
);
assert_eq!(
aparato.instantaneas(),
1,
"la instantánea no está en el repositorio que decidió la serialización"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn abrir_repositorio_inicializa_el_que_no_esta_y_conecta_con_el_que_esta() {
let aparato = Aparato::nuevo().await;
let motor = aparato.motor_con(abridor_local);
let primera = motor
.abrir_repositorio(aparato.acceso_de(&aparato.repositorio).await)
.await
.expect("un destino vacío y alcanzable se inicializa");
assert_eq!(
primera,
PuestaEnMarcha::Inicializado,
"un destino sin repositorio se inicializa, y dijo otra cosa"
);
let segunda = motor
.abrir_repositorio(aparato.acceso_de(&aparato.repositorio).await)
.await
.expect("un destino que ya tiene repositorio se abre");
assert_eq!(
segunda,
PuestaEnMarcha::Conectado,
"un destino que ya tiene repositorio se conecta, y dijo otra cosa"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn listar_snapshots_lee_del_repositorio_y_respeta_la_ventana() {
let aparato = Aparato::nuevo().await;
sembrar_snapshots(
&aparato.destino,
PASSPHRASE_DE_PRUEBA,
&[
"2026-01-01T10:00:00Z",
"2026-01-02T10:00:00Z",
"2026-01-03T10:00:00Z",
],
);
let motor = aparato.motor_con(abridor_local);
let devueltas = motor
.listar_snapshots(aparato.acceso_de(&aparato.repositorio).await, 2)
.await
.expect("el repositorio sembrado se lista");
assert_eq!(
devueltas.len(),
2,
"el repositorio tiene tres instantáneas y `job_executions` está vacía: pedir dos y \
recibir {} dice a la vez que la ventana acota y de dónde sale la lista",
devueltas.len()
);
for resumen in &devueltas {
assert!(
!resumen.id.is_empty() && !resumen.rutas.is_empty() && resumen.etiqueta.is_some(),
"el resumen no trae lo que pide el panel: {resumen:?}"
);
}
assert!(
devueltas.windows(2).all(|par| par[0].fecha >= par[1].fecha),
"el listado no viene de la más reciente a la más antigua: {:?}",
devueltas
.iter()
.map(|resumen| resumen.fecha)
.collect::<Vec<_>>()
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn abrir_repositorio_de_s3_desde_el_runtime_no_entra_en_panico() {
let aparato = Aparato::nuevo().await;
let endpoint = destino_de_s3_que_rechaza();
let repositorio = aparato
.dar_de_alta_el_repositorio_con_endpoint("destino-que-rechaza", &endpoint)
.await;
let motor = aparato.motor_y_ajustes(Ajustes::nuevos(SOCKET_QUE_NO_EXISTE));
let acceso = aparato.acceso_de(&repositorio).await;
let error = motor
.abrir_repositorio(acceso)
.await
.expect_err("un destino que rechaza no pone nada en marcha");
let (causa, motivo) = match error {
ErrorDelMotor::Repositorio { causa, motivo } => (causa, motivo),
otro => panic!(
"el fallo de la biblioteca tiene que llegar como `ErrorDelMotor::Repositorio` y \
llegó como {otro:?}"
),
};
assert!(
matches!(causa, Error::DestinoNoAlcanzable { .. }),
"el destino que rechaza tiene que llegar como un fallo de destino y no como una \
passphrase que no abre: {causa:?}"
);
assert!(
motivo.contains(&endpoint),
"el motivo tiene que nombrar el destino al que no se llegó: {motivo}"
);
assert!(
!motivo.contains(CLAVE_SECRETA_DE_PRUEBA) && !motivo.contains(PASSPHRASE_DE_PRUEBA),
"el motivo de la fachada publica un secreto del repositorio: {motivo}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn un_panico_del_hilo_de_trabajo_llega_como_error_y_no_se_propaga() {
let aparato = Aparato::nuevo().await;
let motor = aparato.motor_con(abridor_que_panica);
let acceso = aparato.acceso_de(&aparato.repositorio).await;
let error = motor
.abrir_repositorio(acceso)
.await
.expect_err("un hilo que se cae con un pánico no puede dar un repositorio abierto");
let motivo = match error {
ErrorDelMotor::TrabajoInterrumpido { motivo } => motivo,
otro => panic!(
"el pánico del hilo de trabajo tiene que llegar como `TrabajoInterrumpido` y llegó \
como {otro:?}"
),
};
assert!(
motivo.contains("pánico"),
"el motivo tiene que decir que el hilo se cayó con un pánico: {motivo}"
);
assert!(
!motivo.contains("se cae a propósito"),
"el motivo publica el texto del pánico, que es arbitrario y no puede salir hacia \
fuera: {motivo}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn comprobar_dice_que_el_motor_de_contenedores_no_contesta() {
let aparato = Aparato::nuevo().await;
let motor = aparato.motor_con(abridor_local);
let error = motor
.comprobar()
.await
.expect_err("un socket que no existe no puede contestar al latido");
assert!(
matches!(error, ErrorDelMotor::MotorDeContenedores { .. }),
"el fallo del motor de contenedores no llega como tal: {error:?}"
);
}
async fn instalar_el_binario_del_volcado(contenedor: &str, escritura: &str) {
let cliente = crate::test_support::cliente_del_motor()
.expect("estas pruebas necesitan un motor de contenedores");
let orden = crate::docker::Orden {
cmd: vec!["sh".to_owned(), "-c".to_owned(), escritura.to_owned()],
env: Vec::new(),
};
let codigo = cliente
.ejecutar(
contenedor,
&orden,
crate::test_support::TIMEOUT_DEL_MOTOR,
|_trozo| {},
)
.await
.expect("el alta del binario dentro del contenedor no puede fallar");
assert_eq!(codigo, 0, "el guion que prepara el contenedor falló");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "requiere un motor de contenedores, y el CI no levanta ninguno: se ejecuta con cargo test -- --ignored"]
async fn el_motor_copia_un_volcado_de_punta_a_punta() {
const GUION: &str = r##"printf "#!/bin/sh\nseq 1 200000\n" > /usr/local/bin/mariadb-dump; chmod +x /usr/local/bin/mariadb-dump"##;
let contenedor = crate::test_support::ContenedorDePrueba::nuevo(
"el_motor_copia_un_volcado_de_punta_a_punta",
)
.await;
instalar_el_binario_del_volcado(contenedor.nombre(), GUION).await;
let aparato = Aparato::nuevo().await;
let target = aparato
.target_de_contenedor(contenedor.nombre(), None)
.await;
let tarea = aparato.tarea_normal(target).await;
let socket = crate::test_support::socket_del_motor()
.expect("estas pruebas necesitan un socket de motor de contenedores");
let motor = aparato.motor_y_ajustes(Ajustes::nuevos(socket).con_abridor(abridor_local));
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
let fila = esperar_el_final_con(
&aparato.dependencias.ejecuciones,
&ejecucion.id,
PLAZO_DEL_CONTENEDOR,
)
.await;
assert_eq!(
fila.status,
Estado::Success,
"la copia de un volcado de verdad tenía que acabar bien: {:?}",
fila.error_message
);
let instantanea = fila
.snapshot_id
.clone()
.expect("una copia buena deja su instantánea");
let esperado: Vec<u8> = (1..=200_000)
.map(|numero| format!("{numero}\n"))
.collect::<String>()
.into_bytes();
let guardado = volcado_guardado(
&aparato.destino,
PASSPHRASE_DE_PRUEBA,
&instantanea,
"tienda.sql",
);
assert_eq!(
guardado.as_deref(),
Some(esperado.as_slice()),
"lo que quedó dentro de la instantánea no es lo que el contenedor escribió"
);
assert_eq!(
fila.bytes_processed,
esperado.len() as i64,
"los bytes procesados no son los que salieron del `exec`"
);
contenedor.retirar().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "requiere un motor de contenedores, y el CI no levanta ninguno: se ejecuta con cargo test -- --ignored"]
async fn el_motivo_de_un_fallo_que_repite_la_contrasena_no_la_publica() {
const CONTRASENA: &str = "Zq7-clave-de-base-distinguible-motor";
const GUION: &str = r##"printf "#!/bin/sh\nprintf 'no puedo con %%s' \"\$MYSQL_PWD\" >&2\nexit 1\n" > /usr/local/bin/mariadb-dump; chmod +x /usr/local/bin/mariadb-dump"##;
let contenedor = crate::test_support::ContenedorDePrueba::nuevo(
"el_motivo_de_un_fallo_que_repite_la_contrasena",
)
.await;
instalar_el_binario_del_volcado(contenedor.nombre(), GUION).await;
let aparato = Aparato::nuevo().await;
let target = aparato
.target_de_contenedor(contenedor.nombre(), Some(CONTRASENA))
.await;
let tarea = aparato.tarea_normal(target).await;
let socket = crate::test_support::socket_del_motor()
.expect("estas pruebas necesitan un socket de motor de contenedores");
let motor = aparato.motor_y_ajustes(Ajustes::nuevos(socket).con_abridor(abridor_local));
let ejecucion = motor.encolar(tarea).await.expect("tiene que entrar");
let fila = esperar_el_final_con(
&aparato.dependencias.ejecuciones,
&ejecucion.id,
PLAZO_DEL_CONTENEDOR,
)
.await;
assert_eq!(
fila.status,
Estado::Failed,
"un contenedor que sale con uno no puede dar una copia buena"
);
let motivo = fila.error_message.clone().unwrap_or_default();
assert!(
!motivo.contains(CONTRASENA),
"el motivo que se guarda publica la contraseña de la base: {motivo}"
);
for longitud in 1..=CONTRASENA.len() {
assert!(
!motivo.contains(&CONTRASENA[..longitud]),
"el motivo guarda los primeros {longitud} caracteres de la contraseña: {motivo}"
);
}
assert!(
motivo.contains(crate::secret::REDACTED),
"el motivo no dice que haya algo redactado: {motivo}"
);
assert!(
motivo.contains('1'),
"el motivo no dice con qué código salió el comando de extracción: {motivo}"
);
assert!(
fila.finished_at.is_some(),
"una ejecución fallida tiene que llevar su `finished_at`"
);
contenedor.retirar().await;
}
}