use uuid::Uuid;
use crate::db::entities::Ejecucion;
use crate::engine::ErrorDelMotor;
use super::planificacion::Planificador;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Decision {
Seguir,
Saltar,
Retirar,
Parar,
}
pub(crate) fn decidir(desenlace: &Result<Ejecucion, ErrorDelMotor>) -> Decision {
match desenlace {
Ok(_) => Decision::Seguir,
Err(ErrorDelMotor::YaTieneUnaEnCurso { .. })
| Err(ErrorDelMotor::ColaDelRepositorioLlena { .. })
| Err(ErrorDelMotor::SinQuienAtiendaLaCola { .. }) => Decision::Saltar,
Err(ErrorDelMotor::TareaQueNoEsta { .. }) => Decision::Retirar,
Err(ErrorDelMotor::MotorParado) => Decision::Parar,
Err(ErrorDelMotor::Base { .. }) => Decision::Seguir,
Err(_) => Decision::Seguir,
}
}
impl Planificador {
pub(crate) async fn disparar(&self, job_id: Uuid) {
if !self.0.sigue_abierta() {
return;
}
let mut estado = self.0.estado.lock().await;
if !self.0.sigue_abierta() {
return;
}
let desenlace = self.0.motor.encolar(job_id).await;
match decidir(&desenlace) {
Decision::Seguir => {
if let Err(error) = &desenlace {
tracing::warn!(
job_id = %job_id,
%error,
"el disparo no ha encolado, pero se sigue: el siguiente tick volverá a intentarlo"
);
}
}
Decision::Saltar => {
if let Err(error) = &desenlace {
tracing::warn!(
job_id = %job_id,
%error,
"el disparo se salta este tick"
);
}
}
Decision::Retirar => {
tracing::warn!(
job_id = %job_id,
"la tarea del disparo ya no existe: se retira de la planificación"
);
if let Some(mango) = estado.planificadas.get(&job_id).copied() {
let reloj = self.0.reloj.lock().await;
if let Err(error) = estado.aplicar_retirada(job_id, reloj.remove(&mango).await)
{
tracing::warn!(
job_id = %job_id,
%error,
"el mango de la tarea retirada no se ha podido quitar del reloj: se \
reintentará en la siguiente reconciliación"
);
}
}
}
Decision::Parar => {
tracing::error!(
job_id = %job_id,
"el motor ha dejado de aceptar trabajo: el planificador se para"
);
self.0.cerrar_la_puerta();
}
}
}
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use std::time::{Duration, Instant};
use time::OffsetDateTime;
use tokio::sync::broadcast::error::TryRecvError;
use uuid::Uuid;
use super::{Decision, decidir};
use crate::db::DbError;
use crate::db::entities::{Ejecucion, Estado};
use crate::engine::{ErrorDelMotor, EventoDeEjecucion};
use crate::scheduler::pruebas::{Aparato, CRON_DE_CINCO};
const PLAZO_DE_ESPERA: Duration = Duration::from_secs(30);
async fn esperar_el_final(aparato: &Aparato, tarea: &Uuid) {
let limite = Instant::now() + PLAZO_DE_ESPERA;
loop {
let estado = aparato.estado_de_la_ejecucion(tarea).await;
if estado == "success" || estado == "failed" {
return;
}
assert!(
Instant::now() < limite,
"la ejecución de la prueba no llegó a un estado terminal en {PLAZO_DE_ESPERA:?}: \
{estado}"
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
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
}
fn nacimiento<'a>(eventos: &'a [EventoDeEjecucion], tarea: &Uuid) -> &'a EventoDeEjecucion {
match eventos
.iter()
.find(|evento| evento.job_id == *tarea && evento.status == Estado::Pending)
{
Some(evento) => evento,
None => panic!("no se publicó el nacimiento de la tarea {tarea}: llegaron {eventos:?}"),
}
}
#[test]
fn la_tabla_de_desenlaces_reparte_las_salidas_de_encolar() {
let tarea = Uuid::new_v4();
let repositorio = Uuid::new_v4();
let nacida = Ejecucion {
id: Uuid::new_v4(),
job_id: tarea,
status: Estado::Pending,
snapshot_id: None,
bytes_processed: 0,
bytes_uploaded: 0,
error_message: None,
started_at: OffsetDateTime::now_utc(),
finished_at: None,
};
let tabla = [
(Ok(nacida), Decision::Seguir),
(
Err(ErrorDelMotor::YaTieneUnaEnCurso { job_id: tarea }),
Decision::Saltar,
),
(
Err(ErrorDelMotor::ColaDelRepositorioLlena {
repository_id: repositorio,
}),
Decision::Saltar,
),
(
Err(ErrorDelMotor::SinQuienAtiendaLaCola {
repository_id: repositorio,
motivo: "el hilo de la cola no se pudo crear".to_owned(),
}),
Decision::Saltar,
),
(
Err(ErrorDelMotor::TareaQueNoEsta { job_id: tarea }),
Decision::Retirar,
),
(Err(ErrorDelMotor::MotorParado), Decision::Parar),
(
Err(ErrorDelMotor::Base {
causa: DbError::DirectorioAusente {
ruta: PathBuf::from("/no/existe/kesto"),
},
}),
Decision::Seguir,
),
];
for (desenlace, esperada) in &tabla {
assert_eq!(
decidir(desenlace),
*esperada,
"el desenlace {desenlace:?} no se reparte como manda el requisito 5"
);
}
}
#[tokio::test]
async fn una_tarea_ya_en_curso_se_salta_y_no_crea_una_segunda_ejecucion() {
let aparato = Aparato::nuevo().await;
let tarea = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
aparato.dejar_una_en_curso(&tarea).await;
planificador.disparar(tarea).await;
assert_eq!(
aparato.ejecuciones_de(&tarea).await,
1,
"un disparo sobre una tarea que ya está en curso creó una segunda ejecución"
);
assert!(
!planificador.esta_parado(),
"un `409` no puede parar al planificador"
);
assert!(
planificador.planificadas().contains(&tarea),
"una tarea saltada dejó de estar planificada: el siguiente tick no podría reintentarla"
);
}
#[tokio::test]
async fn una_tarea_borrada_se_retira_de_la_planificacion_al_disparar() {
let aparato = Aparato::nuevo().await;
let tarea = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
assert!(
aparato
.tareas
.borrar(&tarea)
.await
.expect("el borrado de la tarea no puede fallar"),
"la tarea de la prueba tenía que existir para borrarse"
);
planificador.disparar(tarea).await;
assert!(
!planificador.planificadas().contains(&tarea),
"una tarea borrada siguió planificada tras el disparo: {:?}",
planificador.planificadas()
);
assert_eq!(
aparato.ejecuciones_de(&tarea).await,
0,
"el disparo de una tarea borrada creó una ejecución"
);
assert!(
!planificador.esta_parado(),
"un `404` no puede parar al planificador"
);
}
#[tokio::test]
async fn un_motor_parado_detiene_el_planificador() {
let aparato = Aparato::nuevo().await;
let tarea = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
aparato.motor.parar(PLAZO_DE_ESPERA);
planificador.disparar(tarea).await;
assert!(
planificador.esta_parado(),
"un motor parado no detuvo al planificador: seguiría intentando encolar en cada tick"
);
assert_eq!(
aparato.ejecuciones_de(&tarea).await,
0,
"un disparo contra un motor parado creó una ejecución"
);
}
#[tokio::test]
async fn un_error_de_base_en_un_disparo_no_para_el_planificador() {
let aparato = Aparato::nuevo().await;
let tarea = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
aparato.pool.close().await;
planificador.disparar(tarea).await;
assert!(
!planificador.esta_parado(),
"un fallo de base en un tick paró al planificador: un bloqueo pasajero dejaría el reloj \
caído"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn llegada_la_hora_la_tarea_queda_con_una_ejecucion_en_pending() {
let aparato = Aparato::nuevo().await;
let ocupante = aparato.tarea(CRON_DE_CINCO, true).await;
let tarea = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
aparato
.motor
.encolar(ocupante)
.await
.expect("el ocupante tiene que entrar en su cola");
planificador.disparar(tarea).await;
assert_eq!(
aparato.ejecuciones_de(&tarea).await,
1,
"el disparo no dejó una fila nueva en `job_executions`"
);
assert_eq!(
aparato.estado_de_la_ejecucion(&tarea).await,
"pending",
"la ejecución del disparo no nació en `pending`"
);
esperar_el_final(&aparato, &ocupante).await;
esperar_el_final(&aparato, &tarea).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_origen_de_la_ejecucion_no_es_distinguible_en_la_base() {
let aparato = Aparato::nuevo().await;
let ocupante = aparato.tarea(CRON_DE_CINCO, true).await;
let del_reloj = aparato.tarea(CRON_DE_CINCO, true).await;
let a_mano = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
aparato
.motor
.encolar(ocupante)
.await
.expect("el ocupante tiene que entrar en su cola");
planificador.disparar(del_reloj).await;
aparato
.motor
.encolar(a_mano)
.await
.expect("el disparo manual tiene que entrar");
let columnas = aparato.columnas_de_ejecuciones().await;
assert_eq!(
columnas,
[
"id",
"job_id",
"status",
"snapshot_id",
"bytes_processed",
"bytes_uploaded",
"error_message",
"started_at",
"finished_at",
],
"la tabla `job_executions` no tiene las columnas del §3: una de más podría estar \
diciendo de dónde vino la ejecución"
);
let del_reloj_fila = aparato.ejecucion_de(&del_reloj).await;
let a_mano_fila = aparato.ejecucion_de(&a_mano).await;
assert_eq!(
del_reloj_fila.status,
Estado::Pending,
"la ejecución del reloj no nació en `pending`"
);
assert_eq!(del_reloj_fila.status, a_mano_fila.status);
assert_eq!(del_reloj_fila.snapshot_id, a_mano_fila.snapshot_id);
assert_eq!(del_reloj_fila.bytes_processed, a_mano_fila.bytes_processed);
assert_eq!(del_reloj_fila.bytes_uploaded, a_mano_fila.bytes_uploaded);
assert_eq!(del_reloj_fila.error_message, a_mano_fila.error_message);
assert_eq!(del_reloj_fila.finished_at, a_mano_fila.finished_at);
assert_ne!(
del_reloj_fila.id, a_mano_fila.id,
"las dos ejecuciones de la prueba son la misma fila"
);
esperar_el_final(&aparato, &ocupante).await;
esperar_el_final(&aparato, &del_reloj).await;
esperar_el_final(&aparato, &a_mano).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_nacimiento_se_publica_en_el_hub_como_el_de_un_disparo_manual() {
let aparato = Aparato::nuevo().await;
let del_reloj = aparato.tarea(CRON_DE_CINCO, true).await;
let a_mano = aparato.tarea(CRON_DE_CINCO, true).await;
let planificador = aparato.planificador().await;
let mut receptor = aparato
.motor
.hub()
.suscribir()
.expect("el hub no está cerrado en esta prueba");
planificador.disparar(del_reloj).await;
let eventos_del_reloj = recoger(&mut receptor);
let manual = aparato
.motor
.encolar(a_mano)
.await
.expect("el disparo manual tiene que entrar");
let eventos_a_mano = recoger(&mut receptor);
let nacimiento_del_reloj = nacimiento(&eventos_del_reloj, &del_reloj).clone();
let nacimiento_a_mano = nacimiento(&eventos_a_mano, &a_mano).clone();
assert_eq!(
nacimiento_a_mano.execution_id, manual.id,
"el evento del disparo manual no es el de la ejecución que se encoló"
);
assert_eq!(
nacimiento_del_reloj.status,
Estado::Pending,
"el nacimiento del disparo del reloj no llegó en `pending`"
);
assert!(
!nacimiento_del_reloj.current_action.is_empty(),
"el nacimiento del disparo del reloj no dice qué está haciendo la ejecución"
);
assert_eq!(
nacimiento_del_reloj,
EventoDeEjecucion {
execution_id: nacimiento_del_reloj.execution_id,
job_id: nacimiento_del_reloj.job_id,
..nacimiento_a_mano
},
"el nacimiento del disparo del reloj no trae los campos de uno manual: haría falta un \
evento nuevo, y el requisito lo prohíbe"
);
esperar_el_final(&aparato, &del_reloj).await;
esperar_el_final(&aparato, &a_mano).await;
}
}