use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};
use time::OffsetDateTime;
use uuid::Uuid;
use super::extraccion::{
CAPACIDAD_DEL_CANAL, ErrorDeExtraccion, INACTIVIDAD_MAXIMA, Objetivo, Resultado, Ruta,
motivo_para_guardar, recortar, ruta, volcar,
};
use super::hub::{EventoDeEjecucion, Hub};
use super::{Ajustes, Contexto};
use crate::db::entities::{Ejecucion, Estado};
use crate::docker::{Cliente, Orden};
use crate::rustic::{
Acceso, AvanceDelVolcado, AvisoDeProgreso, BarrasDelMotor, Copia, CopiaHecha,
ProductorDelVolcado, PuestaEnMarcha, Repositorio, Retencion, TunelDelVolcado,
tunel_del_volcado,
};
pub(crate) const ACCION_ENCOLADA: &str = "encolada";
const ACCION_LEYENDO_LA_FILA: &str = "leyendo la tarea, el target y el repositorio";
const ACCION_ABRIENDO_EL_REPOSITORIO: &str = "abriendo el repositorio";
const ACCION_EXTRAYENDO_EL_VOLCADO: &str = "extrayendo el volcado";
const ACCION_ARCHIVANDO: &str = "archivando";
const ACCION_RETENIENDO: &str = "aplicando la retención";
pub(crate) const ACCION_TERMINANDO: &str = "terminando";
const HOST_DEL_MOTOR: &str = "kesto";
const NOMBRE_DE_VOLCADO_POR_DEFECTO: &str = "volcado";
const UMBRAL_DE_AVISO: u64 = 1024 * 1024;
const INTERVALO_MINIMO_DE_AVISO: Duration = Duration::from_secs(1);
pub(crate) fn ejecutar(contexto: &Contexto, fila: &Ejecucion, repository_id: Uuid) {
let en_curso = match marcar_en_curso(contexto, fila) {
Ok(Some(en_curso)) => en_curso,
Ok(None) => {
tracing::warn!(
execution_id = %fila.id,
"la fila de la ejecución ya no estaba al ir a empezar: no hay nada que ejecutar"
);
return;
}
Err(causa) => {
tracing::error!(
execution_id = %fila.id,
%causa,
"no se ha podido pasar la ejecución a `running`"
);
let motivo = recortar(&format!(
"la ejecución no ha llegado a empezar: no se ha podido pasar a `running`: {causa}"
));
fallar_si_no_termino(contexto, &fila.id, &motivo);
return;
}
};
let en_vivo = Arc::new(EnVivo::nueva(contexto, &en_curso));
en_vivo.accion(ACCION_LEYENDO_LA_FILA);
en_vivo.publicar();
let cierre = correr(contexto, &en_curso, &en_vivo, repository_id);
en_vivo.accion(ACCION_TERMINANDO);
escribir_el_final(contexto, &en_curso, &cierre, en_vivo.cifras());
contexto
.hub
.publicar(en_vivo.evento(cierre.estado, ACCION_TERMINANDO));
}
fn escribir_el_final(contexto: &Contexto, fila: &Ejecucion, cierre: &Cierre, cifras: Cifras) {
let escrito = contexto.en_la_base(
contexto
.dependencias
.ejecuciones
.actualizar(&la_fila_final(fila, cierre, cifras)),
);
constancia_del_final(escrito, &fila.id);
}
async fn escribir_el_final_desde_el_runtime(
contexto: &Contexto,
fila: &Ejecucion,
cierre: &Cierre,
cifras: Cifras,
) {
let escrito = contexto
.dependencias
.ejecuciones
.actualizar(&la_fila_final(fila, cierre, cifras))
.await;
constancia_del_final(escrito, &fila.id);
}
fn la_fila_final(fila: &Ejecucion, cierre: &Cierre, cifras: Cifras) -> Ejecucion {
let mut final_de_la_fila = fila.clone();
final_de_la_fila.status = cierre.estado;
final_de_la_fila.snapshot_id = cierre.snapshot_id.clone();
final_de_la_fila.error_message = cierre.motivo.clone();
final_de_la_fila.bytes_processed = como_entero(cifras.procesados);
final_de_la_fila.bytes_uploaded = como_entero(cifras.subidos);
final_de_la_fila.finished_at = Some(OffsetDateTime::now_utc());
final_de_la_fila
}
fn constancia_del_final(escrito: Result<bool, crate::db::DbError>, id: &Uuid) {
match escrito {
Ok(true) => {}
Ok(false) => tracing::warn!(
execution_id = %id,
"la ejecución ha terminado y su fila ya no estaba: no hay dónde escribir el final"
),
Err(causa) => tracing::error!(
execution_id = %id,
%causa,
"no se ha podido escribir el final de la ejecución"
),
}
}
pub(crate) fn fallar_si_no_termino(contexto: &Contexto, id: &Uuid, motivo: &str) {
contexto.en_la_base(marcar_fallida_si_no_termino(contexto, id, motivo));
}
pub(crate) async fn marcar_fallida_si_no_termino(contexto: &Contexto, id: &Uuid, motivo: &str) {
let fila = match contexto.dependencias.ejecuciones.leer(id).await {
Ok(Some(fila)) => fila,
Ok(None) => {
tracing::warn!(%id, "la ejecución que se quería marcar como fallida ya no está");
return;
}
Err(causa) => {
tracing::error!(
%id,
%causa,
"no se ha podido leer la ejecución para marcarla como fallida"
);
return;
}
};
if matches!(fila.status, Estado::Success | Estado::Failed) {
return;
}
escribir_el_final_desde_el_runtime(
contexto,
&fila,
&Cierre::fallida(recortar(motivo)),
Cifras::nada(),
)
.await;
}
fn marcar_en_curso(
contexto: &Contexto,
fila: &Ejecucion,
) -> Result<Option<Ejecucion>, crate::db::DbError> {
let mut en_curso = fila.clone();
en_curso.status = Estado::Running;
if !contexto.en_la_base(contexto.dependencias.ejecuciones.actualizar(&en_curso))? {
return Ok(None);
}
Ok(Some(en_curso))
}
fn correr(
contexto: &Contexto,
fila: &Ejecucion,
en_vivo: &Arc<EnVivo>,
repository_id: Uuid,
) -> Cierre {
let ajustes = &contexto.ajustes;
let tarea = match leer_la_tarea(contexto, &fila.job_id) {
Ok(tarea) => tarea,
Err(motivo) => return Cierre::fallida(motivo),
};
let target = match leer_el_target(contexto, &tarea.target_id) {
Ok(target) => target,
Err(motivo) => return Cierre::fallida(motivo),
};
let repositorio = match leer_el_repositorio(contexto, &repository_id) {
Ok(repositorio) => repositorio,
Err(motivo) => return Cierre::fallida(motivo),
};
let objetivo = Objetivo::from(&target);
let por_donde = match ruta(&objetivo) {
Ok(por_donde) => por_donde,
Err(causa) => return Cierre::fallida(recortar(&causa.to_string())),
};
let acceso = Acceso {
endpoint: repositorio.endpoint.clone(),
bucket: repositorio.bucket.clone(),
region: repositorio.region.clone(),
access_key_id: repositorio.access_key_id.clone(),
secret_access_key: repositorio.secret_access_key.clone(),
passphrase: repositorio.passphrase.clone(),
};
let nombre = nombre_del_volcado(&target.name);
let etiqueta = fila.job_id.hyphenated().to_string();
let retencion = Retencion {
keep_daily: tarea.keep_daily,
keep_weekly: tarea.keep_weekly,
keep_monthly: tarea.keep_monthly,
};
en_vivo.accion(ACCION_ABRIENDO_EL_REPOSITORIO);
en_vivo.publicar();
let barras = en_vivo.barras_de_esta_ejecucion();
let (copia_del_repositorio, puesta_en_marcha) = match (ajustes.abridor)(&acceso, barras) {
Ok(abierto) => abierto,
Err(causa) => {
return Cierre::fallida(motivo_de_la_biblioteca(&causa.to_string(), &acceso));
}
};
match puesta_en_marcha {
PuestaEnMarcha::Inicializado => tracing::info!(
endpoint = %acceso.endpoint,
"el destino no tenía repositorio y se ha creado con esta copia"
),
PuestaEnMarcha::Conectado => tracing::info!(
endpoint = %acceso.endpoint,
"el destino ya tenía repositorio y se ha conectado con él"
),
}
let hecho = match por_donde {
Ruta::PorFichero { ruta } => copiar_el_directorio(
&copia_del_repositorio,
&acceso,
&ruta,
&nombre,
&etiqueta,
en_vivo,
),
Ruta::PorContenedor { contenedor, orden } => copiar_el_volcado(
&copia_del_repositorio,
&acceso,
&contenedor,
orden,
&nombre,
&etiqueta,
en_vivo,
ajustes,
),
};
let Ok(hecho) = hecho else {
return hecho.unwrap_err();
};
en_vivo.accion(ACCION_RETENIENDO);
en_vivo.publicar();
if let Err(causa) = copia_del_repositorio.aplicar_retencion(&retencion) {
let motivo = motivo_de_la_biblioteca(&causa.to_string(), &acceso);
tracing::warn!(
%etiqueta,
%motivo,
"la copia ha terminado bien y la retención ha fallado: lo que hay que mirar es la \
retención, no la copia"
);
}
Cierre::hecha(&hecho)
}
fn leer_la_tarea(contexto: &Contexto, job_id: &Uuid) -> Result<crate::db::entities::Tarea, String> {
match contexto.en_la_base(contexto.dependencias.tareas.leer(job_id)) {
Ok(Some(tarea)) => Ok(tarea),
Ok(None) => Err(format!(
"la tarea `{job_id}` ya no está: la ejecución se encoló para una fila que se ha \
borrado entre el encolado y la ejecución"
)),
Err(causa) => Err(recortar(&format!(
"la tarea `{job_id}` no se ha podido leer: {causa}"
))),
}
}
fn leer_el_target(
contexto: &Contexto,
target_id: &Uuid,
) -> Result<crate::db::entities::Target, String> {
match contexto.en_la_base(contexto.dependencias.targets.leer(target_id)) {
Ok(Some(target)) => Ok(target),
Ok(None) => Err(format!(
"el target `{target_id}` de la tarea ya no está: sin él no hay nada que copiar"
)),
Err(causa) => Err(recortar(&format!(
"el target `{target_id}` no se ha podido leer: {causa}"
))),
}
}
fn leer_el_repositorio(
contexto: &Contexto,
repository_id: &Uuid,
) -> Result<crate::db::entities::Repositorio, String> {
match contexto.en_la_base(contexto.dependencias.repositorios.leer(repository_id)) {
Ok(Some(repositorio)) => Ok(repositorio),
Ok(None) => Err(format!(
"el repositorio `{repository_id}` de la tarea ya no está: sin él no hay dónde copiar"
)),
Err(causa) => Err(recortar(&format!(
"el repositorio `{repository_id}` no se ha podido leer: {causa}"
))),
}
}
fn copiar_el_directorio(
repositorio: &crate::rustic::Repositorio,
acceso: &Acceso,
ruta: &Path,
nombre: &str,
etiqueta: &str,
en_vivo: &Arc<EnVivo>,
) -> Result<CopiaHecha, Cierre> {
let copia = Copia {
rutas: vec![ruta.to_path_buf()],
nombre: nombre.to_owned(),
etiqueta: etiqueta.to_owned(),
host: HOST_DEL_MOTOR.to_owned(),
comando: format!("copia del directorio {}", ruta.display()),
};
en_vivo.accion(ACCION_ARCHIVANDO);
en_vivo.publicar();
repositorio
.archivar_directorio(&copia)
.map_err(|causa| Cierre::fallida(motivo_de_la_biblioteca(&causa.to_string(), acceso)))
}
#[allow(clippy::too_many_arguments)]
fn copiar_el_volcado(
repositorio: &Repositorio,
acceso: &Acceso,
contenedor: &str,
orden: Orden,
nombre: &str,
etiqueta: &str,
en_vivo: &Arc<EnVivo>,
ajustes: &Ajustes,
) -> Result<CopiaHecha, Cierre> {
let tunel = tunel_del_volcado(nombre.to_owned(), CAPACIDAD_DEL_CANAL, INACTIVIDAD_MAXIMA)
.map_err(|causa| Cierre::fallida(recortar(&causa.to_string())))?;
let TunelDelVolcado {
origen,
productor,
entrega,
avance,
} = tunel;
en_vivo.conectar_el_volcado(avance);
let copia = Copia {
rutas: vec![PathBuf::from(nombre)],
nombre: nombre.to_owned(),
etiqueta: etiqueta.to_owned(),
host: HOST_DEL_MOTOR.to_owned(),
comando: orden.cmd.join(" "),
};
en_vivo.accion(ACCION_EXTRAYENDO_EL_VOLCADO);
en_vivo.publicar();
let hilo_del_productor = lanzar_el_productor(
ajustes.socket_del_motor.clone(),
ajustes.espera_del_motor,
contenedor.to_owned(),
orden,
productor,
)
.map_err(|motivo| Cierre::fallida(recortar(&motivo)))?;
en_vivo.accion(ACCION_ARCHIVANDO);
en_vivo.publicar();
let archivado = repositorio.archivar_volcado(&copia, &origen);
let vuelta = hilo_del_productor.join();
let hecho = match archivado {
Ok(hecho) => hecho,
Err(causa) => {
return Err(Cierre::fallida(motivo_de_la_biblioteca(
&causa.to_string(),
acceso,
)));
}
};
let resultado = match vuelta {
Ok(Ok(resultado)) => resultado,
Ok(Err(motivo)) => {
return Err(descartar_copia_fallida(
|| olvidar_la_instantanea(repositorio, &hecho.id),
motivo,
));
}
Err(_) => {
return Err(descartar_copia_fallida(
|| olvidar_la_instantanea(repositorio, &hecho.id),
recortar(
"el hilo que extraía el volcado se ha caído sin dejar hechos: la copia no se \
puede dar por buena",
),
));
}
};
match el_volcado_se_entrego(entrega.completa(), resultado.codigo, hecho.ficheros) {
Ok(()) => Ok(hecho),
Err(fallo) => Err(descartar_copia_fallida(
|| olvidar_la_instantanea(repositorio, &hecho.id),
motivo_del_fallo(fallo, &resultado, hecho.ficheros),
)),
}
}
fn descartar_copia_fallida<F>(olvidar: F, motivo: String) -> Cierre
where
F: FnOnce() -> Result<(), String>,
{
if let Err(fallo_del_olvido) = olvidar() {
tracing::warn!(
motivo = %fallo_del_olvido,
"no se ha podido retirar la instantánea de una copia que no se ha dado por buena: \
queda como residuo hasta la próxima purga"
);
}
Cierre::fallida(motivo)
}
fn olvidar_la_instantanea(repositorio: &Repositorio, id: &str) -> Result<(), String> {
repositorio
.olvidar_instantanea(id)
.map_err(|causa| causa.to_string())
}
fn lanzar_el_productor(
socket: String,
espera: Duration,
contenedor: String,
orden: Orden,
productor: ProductorDelVolcado,
) -> Result<std::thread::JoinHandle<VueltaDelProductor>, String> {
std::thread::Builder::new()
.name(format!("kesto-volcado-{contenedor}"))
.spawn(move || producir(socket, espera, contenedor, orden, productor))
.map_err(|causa| format!("no se ha podido lanzar el hilo del volcado: {causa}"))
}
type VueltaDelProductor = Result<Resultado, String>;
fn producir(
socket: String,
espera: Duration,
contenedor: String,
orden: Orden,
productor: ProductorDelVolcado,
) -> VueltaDelProductor {
let runtime = match tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
{
Ok(runtime) => runtime,
Err(causa) => {
return Err(recortar(&format!(
"no se ha podido crear el runtime del volcado: {causa}"
)));
}
};
runtime.block_on(async move {
let cliente =
Cliente::conectar(&socket, espera).map_err(|causa| recortar(&causa.to_string()))?;
volcar(&cliente, &contenedor, &orden, productor, INACTIVIDAD_MAXIMA)
.await
.map_err(|causa| motivo_del_volcado(&causa, &orden))
})
}
fn como_entero(bytes: u64) -> i64 {
i64::try_from(bytes).unwrap_or(i64::MAX)
}
fn nombre_del_volcado(del_target: &str) -> String {
let saneado: String = del_target
.chars()
.map(|caracter| {
if caracter.is_ascii_alphanumeric() || matches!(caracter, '.' | '-' | '_') {
caracter
} else {
'_'
}
})
.collect();
let base = if saneado.is_empty() {
NOMBRE_DE_VOLCADO_POR_DEFECTO
} else {
&saneado
};
format!("{base}.sql")
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FalloDelVolcado {
NoSeEntregoEntero,
ElProcesoFallo {
codigo: i32,
},
SinElVolcadoDentro,
}
fn el_volcado_se_entrego(
entrega_completa: bool,
codigo: i32,
ficheros: u64,
) -> Result<(), FalloDelVolcado> {
if !entrega_completa {
return Err(FalloDelVolcado::NoSeEntregoEntero);
}
if codigo != 0 {
return Err(FalloDelVolcado::ElProcesoFallo { codigo });
}
if ficheros < 1 {
return Err(FalloDelVolcado::SinElVolcadoDentro);
}
Ok(())
}
fn motivo_del_fallo(fallo: FalloDelVolcado, resultado: &Resultado, ficheros: u64) -> String {
let que_paso = match fallo {
FalloDelVolcado::NoSeEntregoEntero if resultado.codigo != 0 => format!(
"el volcado no se entregó entero —el comando de extracción salió con el código {}—: \
puede estar truncado, y archivarlo daría por buena una copia incompleta",
resultado.codigo
),
FalloDelVolcado::NoSeEntregoEntero => {
"el volcado no se entregó entero: puede estar truncado, y archivarlo daría por buena \
una copia incompleta"
.to_owned()
}
FalloDelVolcado::ElProcesoFallo { codigo } => {
format!("el comando de extracción salió con el código {codigo}")
}
FalloDelVolcado::SinElVolcadoDentro => format!(
"la instantánea no tiene el volcado dentro: el archivado contó {ficheros} ficheros"
),
};
let lo_que_dijo = resultado.motivo_del_contenedor.trim();
if lo_que_dijo.is_empty() {
return recortar(&que_paso);
}
recortar(&format!("{que_paso}; el contenedor dijo: {lo_que_dijo}"))
}
fn motivo_del_volcado(causa: &ErrorDeExtraccion, orden: &Orden) -> String {
motivo_para_guardar(causa.to_string().into_bytes(), orden)
}
pub(super) fn motivo_de_la_biblioteca(texto: &str, acceso: &Acceso) -> String {
let a_quitar = Orden {
cmd: Vec::new(),
env: vec![
format!("secret_access_key={}", acceso.secret_access_key.expose()),
format!("passphrase={}", acceso.passphrase.expose()),
],
};
motivo_para_guardar(texto.as_bytes().to_vec(), &a_quitar)
}
struct Cierre {
estado: Estado,
snapshot_id: Option<String>,
motivo: Option<String>,
}
impl Cierre {
fn hecha(hecho: &CopiaHecha) -> Self {
Self {
estado: Estado::Success,
snapshot_id: Some(hecho.id.clone()),
motivo: None,
}
}
fn fallida(motivo: String) -> Self {
Self {
estado: Estado::Failed,
snapshot_id: None,
motivo: Some(motivo),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq)]
struct Cifras {
procesados: u64,
subidos: u64,
percentage: Option<f64>,
}
impl Cifras {
const fn nada() -> Self {
Self {
procesados: 0,
subidos: 0,
percentage: None,
}
}
}
#[derive(Debug)]
struct Avance {
total: Option<u64>,
subidos: u64,
ultimo_aviso: Instant,
bytes_del_ultimo_aviso: u64,
}
struct EnVivo {
hub: Hub,
execution_id: Uuid,
job_id: Uuid,
volcado: Mutex<Option<AvanceDelVolcado>>,
accion: Mutex<&'static str>,
avance: Mutex<Avance>,
}
impl EnVivo {
fn nueva(contexto: &Contexto, fila: &Ejecucion) -> Self {
Self {
hub: contexto.hub.clone(),
execution_id: fila.id,
job_id: fila.job_id,
volcado: Mutex::new(None),
accion: Mutex::new(ACCION_ENCOLADA),
avance: Mutex::new(Avance {
total: None,
subidos: 0,
ultimo_aviso: Instant::now(),
bytes_del_ultimo_aviso: 0,
}),
}
}
fn barras_de_esta_ejecucion(self: &Arc<Self>) -> BarrasDelMotor {
let en_vivo = Arc::clone(self);
BarrasDelMotor::nuevas(Arc::new(move |aviso: AvisoDeProgreso| {
en_vivo.al_avisar(aviso);
}))
}
fn conectar_el_volcado(&self, avance: AvanceDelVolcado) {
*candado(&self.volcado) = Some(avance);
}
fn accion(&self, accion: &'static str) {
*candado(&self.accion) = accion;
}
fn cifras(&self) -> Cifras {
let avance = candado(&self.avance);
let del_volcado = candado(&self.volcado).clone();
let subidos = avance.subidos;
let total = avance.total;
let procesados = match del_volcado {
Some(volcado) => volcado.bytes(),
None => total.unwrap_or(subidos),
};
let percentage = match total {
Some(total) if total > 0 => {
Some((100.0 * subidos as f64 / total as f64).clamp(0.0, 100.0))
}
_ => None,
};
Cifras {
procesados,
subidos,
percentage,
}
}
fn evento(&self, estado: Estado, accion: &str) -> EventoDeEjecucion {
let cifras = self.cifras();
EventoDeEjecucion {
execution_id: self.execution_id,
job_id: self.job_id,
status: estado,
bytes_processed: cifras.procesados,
bytes_uploaded: cifras.subidos,
percentage: cifras.percentage,
current_action: accion.to_owned(),
}
}
fn publicar(&self) {
let accion = *candado(&self.accion);
self.hub.publicar(self.evento(Estado::Running, accion));
}
fn al_avisar(&self, aviso: AvisoDeProgreso) {
let toca = {
let mut avance = candado(&self.avance);
match aviso {
AvisoDeProgreso::TotalDeBytes(total) => avance.total = Some(total),
AvisoDeProgreso::Bytes(bytes) => {
avance.subidos = avance.subidos.saturating_add(bytes);
}
AvisoDeProgreso::FinDeBytes => {}
}
let nuevos = avance.subidos > avance.bytes_del_ultimo_aviso;
let por_bytes =
avance.subidos.saturating_sub(avance.bytes_del_ultimo_aviso) >= UMBRAL_DE_AVISO;
let por_tiempo = nuevos && avance.ultimo_aviso.elapsed() >= INTERVALO_MINIMO_DE_AVISO;
if por_bytes || por_tiempo {
avance.ultimo_aviso = Instant::now();
avance.bytes_del_ultimo_aviso = avance.subidos;
}
por_bytes || por_tiempo
};
if toca {
self.publicar();
}
}
}
fn candado<T>(candado: &Mutex<T>) -> MutexGuard<'_, T> {
match candado.lock() {
Ok(guardado) => guardado,
Err(veneno) => veneno.into_inner(),
}
}
#[cfg(test)]
mod tests {
use std::cell::Cell;
use super::{
FalloDelVolcado, descartar_copia_fallida, el_volcado_se_entrego, nombre_del_volcado,
};
use crate::db::entities::Estado;
#[test]
fn la_condicion_de_exito_del_volcado_no_se_hereda_del_archivado() {
assert_eq!(
el_volcado_se_entrego(true, 0, 1),
Ok(()),
"un volcado entregado entero, con el proceso en cero y con su nodo en la instantánea \
es una copia buena"
);
assert_eq!(
el_volcado_se_entrego(false, 0, 1),
Err(FalloDelVolcado::NoSeEntregoEntero),
"un volcado que no se entregó entero se dio por bueno: el archivado silencia el error \
del lector y devuelve `Ok`, así que un `Ok` no puede ser la condición de éxito"
);
assert_eq!(
el_volcado_se_entrego(true, 3, 1),
Err(FalloDelVolcado::ElProcesoFallo { codigo: 3 }),
"un proceso de extracción que sale con 3 no puede dar una copia buena"
);
assert_eq!(
el_volcado_se_entrego(true, 0, 0),
Err(FalloDelVolcado::SinElVolcadoDentro),
"una instantánea sin el volcado dentro no es una copia: el archivado descartó la \
entrada que falló y guardó el árbol sin ella"
);
}
#[test]
fn un_volcado_vacio_con_su_nodo_dentro_es_una_copia_buena() {
assert_eq!(
el_volcado_se_entrego(true, 0, 1),
Ok(()),
"un volcado de cero bytes sigue siendo un nodo dentro de la instantánea, y `bytes` \
vale cero igual que en el truncado: por eso no sirve para decidir"
);
}
#[test]
fn el_nombre_del_volcado_se_sanea_a_partir_del_target() {
assert_eq!(
nombre_del_volcado("tienda"),
"tienda.sql",
"un nombre normal no se toca más allá de la extensión"
);
assert_eq!(
nombre_del_volcado("tienda-de-ropa_2026.v2"),
"tienda-de-ropa_2026.v2.sql",
"los caracteres que sí valen en un nombre de nodo no se sustituyen"
);
assert_eq!(
nombre_del_volcado("volcados/tienda"),
"volcados_tienda.sql",
"un separador dentro del nombre partiría la entrada en trozos"
);
assert_eq!(
nombre_del_volcado("la tienda (madrid)"),
"la_tienda__madrid_.sql",
"lo que no es letra, dígito, punto, guion o subrayado se sustituye"
);
assert_eq!(
nombre_del_volcado(""),
"volcado.sql",
"un nombre vacío sería un pánico dentro de la biblioteca, no un error devuelto"
);
assert_eq!(
nombre_del_volcado(" "),
"___.sql",
"un nombre de solo espacios sobrevive como subrayados: no queda vacío —un camino vacío \
es un pánico dentro de la biblioteca— y no se recorta, que sería cambiar en silencio \
lo que el operador escribió"
);
assert_eq!(
nombre_del_volcado("años"),
"a_os.sql",
"el saneado es ASCII: el nombre acaba en una instantánea que leen otras herramientas"
);
}
#[test]
fn descartar_una_copia_fallida_conserva_el_motivo_original_aunque_falle_el_olvido() {
let intentos = Cell::new(0);
let cierre = descartar_copia_fallida(
|| {
intentos.set(intentos.get() + 1);
Err("el destino no deja retirar la instantánea".to_owned())
},
"el volcado no se entregó entero".to_owned(),
);
assert_eq!(
intentos.get(),
1,
"la instantánea de una copia que no se da por buena tenía que intentar olvidarse una \
vez"
);
assert_eq!(
cierre.estado,
Estado::Failed,
"la copia no se dio por buena y su cierre no es `failed`"
);
assert_eq!(
cierre.snapshot_id, None,
"una copia que no se da por buena no puede publicar un `snapshot_id`"
);
assert_eq!(
cierre.motivo.as_deref(),
Some("el volcado no se entregó entero"),
"el motivo de la retirada sustituyó al del volcado: el operador tiene que leer por qué \
falló la copia, no por qué no se pudo limpiar el residuo —que la purga siguiente se \
llevará—"
);
}
#[test]
fn descartar_una_copia_fallida_es_failed_con_el_motivo_original_cuando_el_olvido_va_bien() {
let cierre =
descartar_copia_fallida(|| Ok(()), "el proceso de extracción salió con 3".to_owned());
assert_eq!(
cierre.estado,
Estado::Failed,
"el olvido fue bien y con eso la copia se coló como buena: sigue sin serlo"
);
assert_eq!(
cierre.snapshot_id, None,
"una copia que no se da por buena no puede publicar un `snapshot_id`"
);
assert_eq!(
cierre.motivo.as_deref(),
Some("el proceso de extracción salió con 3"),
"el motivo del cierre no es el original del volcado"
);
}
}