use std::ffi::OsStr;
use std::io::Read;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender, sync_channel};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use rustic_core::repofile::{Metadata, Node, NodeType};
use rustic_core::{ErrorKind, ReadSource, ReadSourceEntry, RusticError, RusticResult};
use super::Error;
pub struct TunelDelVolcado {
pub origen: OrigenDelVolcado,
pub productor: ProductorDelVolcado,
pub entrega: EntregaDelVolcado,
pub avance: AvanceDelVolcado,
}
#[derive(Clone)]
pub struct EntregaDelVolcado(Arc<AtomicBool>);
impl EntregaDelVolcado {
pub fn completa(&self) -> bool {
self.0.load(Ordering::Acquire)
}
fn marcar(&self) {
self.0.store(true, Ordering::Release);
}
}
impl std::fmt::Debug for EntregaDelVolcado {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("EntregaDelVolcado")
.field("completa", &self.completa())
.finish()
}
}
#[derive(Clone)]
pub struct AvanceDelVolcado(Arc<AtomicU64>);
impl AvanceDelVolcado {
pub fn bytes(&self) -> u64 {
self.0.load(Ordering::Relaxed)
}
fn sumar(&self, bytes: u64) {
self.0.fetch_add(bytes, Ordering::Relaxed);
}
}
impl std::fmt::Debug for AvanceDelVolcado {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AvanceDelVolcado")
.field("bytes", &self.bytes())
.finish()
}
}
pub struct ProductorDelVolcado {
nombre: String,
emisor: SyncSender<Vec<u8>>,
terminado_bien: Arc<AtomicBool>,
avance: AvanceDelVolcado,
}
impl std::fmt::Debug for ProductorDelVolcado {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ProductorDelVolcado")
.field("nombre", &self.nombre)
.finish_non_exhaustive()
}
}
impl ProductorDelVolcado {
pub fn enviar(&self, trozo: Vec<u8>) -> Result<(), Error> {
if trozo.is_empty() {
return Ok(());
}
let bytes = trozo.len() as u64;
self.emisor
.send(trozo)
.map_err(|_| Error::ElArchivadoSeRindio {
nombre: self.nombre.clone(),
})?;
self.avance.sumar(bytes);
Ok(())
}
pub fn terminar(self) {
self.terminado_bien.store(true, Ordering::Release);
}
}
pub fn tunel_del_volcado(
nombre: impl Into<String>,
capacidad: usize,
inactividad_maxima: Duration,
) -> Result<TunelDelVolcado, Error> {
let nombre = nombre.into();
validar_nombre_de_volcado(&nombre)?;
let (emisor, receptor) = sync_channel(capacidad);
let terminado_bien = Arc::new(AtomicBool::new(false));
let entrega = EntregaDelVolcado(Arc::new(AtomicBool::new(false)));
let avance = AvanceDelVolcado(Arc::new(AtomicU64::new(0)));
Ok(TunelDelVolcado {
origen: OrigenDelVolcado {
nombre: nombre.clone(),
receptor: Mutex::new(Some(receptor)),
terminado_bien: Arc::clone(&terminado_bien),
entrega: entrega.clone(),
inactividad_maxima,
},
productor: ProductorDelVolcado {
nombre,
emisor,
terminado_bien,
avance: avance.clone(),
},
entrega,
avance,
})
}
fn validar_nombre_de_volcado(nombre: &str) -> Result<(), Error> {
let sin_nombre = nombre.is_empty() || nombre == "." || nombre == "..";
let es_un_camino = nombre.contains('/') || nombre.contains('\\');
if sin_nombre || es_un_camino {
return Err(Error::NombreDeVolcadoInvalido {
nombre: nombre.to_owned(),
});
}
Ok(())
}
pub struct OrigenDelVolcado {
nombre: String,
receptor: Mutex<Option<Receiver<Vec<u8>>>>,
terminado_bien: Arc<AtomicBool>,
entrega: EntregaDelVolcado,
inactividad_maxima: Duration,
}
impl OrigenDelVolcado {
pub fn nombre(&self) -> &str {
&self.nombre
}
}
impl std::fmt::Debug for OrigenDelVolcado {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OrigenDelVolcado")
.field("nombre", &self.nombre)
.finish_non_exhaustive()
}
}
pub struct LectorDeVolcado {
receptor: Receiver<Vec<u8>>,
sobrante: Vec<u8>,
consumido: usize,
nombre: String,
terminado_bien: Arc<AtomicBool>,
entrega: EntregaDelVolcado,
inactividad_maxima: Duration,
}
impl Read for LectorDeVolcado {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if buf.is_empty() {
return Ok(0);
}
if self.consumido < self.sobrante.len() {
let pendientes = &self.sobrante[self.consumido..];
let a_entregar = pendientes.len().min(buf.len());
buf[..a_entregar].copy_from_slice(&pendientes[..a_entregar]);
self.consumido += a_entregar;
if self.consumido == self.sobrante.len() {
self.sobrante.clear();
self.consumido = 0;
}
return Ok(a_entregar);
}
match self.receptor.recv_timeout(self.inactividad_maxima) {
Ok(datos) => {
let a_entregar = datos.len().min(buf.len());
buf[..a_entregar].copy_from_slice(&datos[..a_entregar]);
if a_entregar < datos.len() {
self.sobrante = datos;
self.consumido = a_entregar;
}
Ok(a_entregar)
}
Err(RecvTimeoutError::Timeout) => Err(self.volcado_estancado()),
Err(RecvTimeoutError::Disconnected) => {
if self.terminado_bien.load(Ordering::Acquire) {
self.entrega.marcar();
Ok(0)
} else {
Err(self.canal_cerrado_sin_final())
}
}
}
}
}
impl LectorDeVolcado {
fn canal_cerrado_sin_final(&self) -> std::io::Error {
std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
format!(
"el productor del volcado `{}` soltó el canal sin declarar el final: el volcado \
puede estar truncado, y archivarlo daría por buena una copia incompleta",
self.nombre
),
)
}
fn volcado_estancado(&self) -> std::io::Error {
std::io::Error::new(
std::io::ErrorKind::TimedOut,
Error::VolcadoEstancado {
nombre: self.nombre.clone(),
inactividad: self.inactividad_maxima,
},
)
}
}
impl ReadSource for OrigenDelVolcado {
type Open = LectorDeVolcado;
type Iter = std::vec::IntoIter<RusticResult<ReadSourceEntry<Self::Open>>>;
fn size(&self) -> RusticResult<Option<u64>> {
Ok(None)
}
fn entries(&self) -> Self::Iter {
let receptor = match self.receptor.lock() {
Ok(mut guardado) => guardado.take(),
Err(veneno) => veneno.into_inner().take(),
};
let Some(receptor) = receptor else {
return vec![Err(error_de_un_solo_uso(&self.nombre))].into_iter();
};
let path = PathBuf::from(&self.nombre);
let node = Node::new_node(
OsStr::new(self.nombre.as_str()),
NodeType::File,
Metadata::default(),
);
let lector = LectorDeVolcado {
receptor,
sobrante: Vec::new(),
consumido: 0,
nombre: self.nombre.clone(),
terminado_bien: Arc::clone(&self.terminado_bien),
entrega: self.entrega.clone(),
inactividad_maxima: self.inactividad_maxima,
};
vec![Ok(ReadSourceEntry {
path,
node,
open: Some(lector),
})]
.into_iter()
}
}
fn error_de_un_solo_uso(nombre: &str) -> Box<RusticError> {
RusticError::new(
ErrorKind::Internal,
format!(
"el origen del volcado `{nombre}` ya entregó su entrada: las entradas de un volcado \
se piden una sola vez, y una segunda llamada no puede devolver un origen vacío"
),
)
}
#[cfg(test)]
mod tests {
use std::io::Read;
use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use rustic_core::ReadSource as _;
use rustic_core::ReadSourceOpen;
use rustic_core::RusticResult;
use rustic_core::repofile::NodeType;
use super::{
Error, LectorDeVolcado, OrigenDelVolcado, ProductorDelVolcado, TunelDelVolcado,
tunel_del_volcado,
};
const PLAZO_DE_PRUEBA: Duration = Duration::from_secs(60);
fn tunel_de_prueba(nombre: &str, capacidad: usize) -> TunelDelVolcado {
tunel_con_plazo(nombre, capacidad, PLAZO_DE_PRUEBA)
}
fn tunel_con_plazo(
nombre: &str,
capacidad: usize,
inactividad_maxima: Duration,
) -> TunelDelVolcado {
tunel_del_volcado(nombre, capacidad, inactividad_maxima)
.expect("el nombre de esta prueba tiene que ser válido")
}
fn nombre_rechazado(tunel: Result<TunelDelVolcado, Error>) -> String {
match tunel {
Err(Error::NombreDeVolcadoInvalido { nombre }) => nombre,
Err(otro) => panic!("ese nombre tenía que rechazarse, y salió otro error: {otro}"),
Ok(_) => panic!("ese nombre tenía que rechazarse"),
}
}
fn lector_de(origen: &OrigenDelVolcado) -> RusticResult<LectorDeVolcado> {
let mut entradas = origen.entries();
let entrada = entradas
.next()
.expect("el origen tiene que dar una entrada")
.expect("la entrada no puede ser un error");
let abierto = entrada
.open
.expect("la entrada tiene que traer cómo abrirse");
abierto.open()
}
#[test]
fn el_tamano_de_un_volcado_no_se_conoce() {
let tunel = tunel_de_prueba("volcado-de-prueba.sql", 4);
let tamano = tunel
.origen
.size()
.expect("preguntar el tamaño no puede fallar");
assert_eq!(
tamano, None,
"un volcado por flujo no puede declarar un total: {tamano:?}"
);
}
#[test]
fn el_lector_entrega_los_bytes_del_productor_en_orden() {
let TunelDelVolcado {
origen, productor, ..
} = tunel_de_prueba("volcado-de-tienda.sql", 2);
let trozos = [
b"CREATE TABLE tienda (id INT);\n".to_vec(),
b"INSERT INTO tienda VALUES (1);\n".to_vec(),
b"INSERT INTO tienda VALUES (2);\n".to_vec(),
];
let esperado: Vec<u8> = trozos.concat();
let total = esperado.len();
let emisor = std::thread::spawn(move || {
for trozo in trozos {
productor
.enviar(trozo)
.expect("el origen sigue vivo, así que el canal no puede estar cerrado");
}
productor.terminar();
});
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let mut leido = Vec::new();
lector
.read_to_end(&mut leido)
.expect("con el final declarado, leer no puede fallar");
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
leido.len(),
total,
"se leyeron {} bytes de {total}",
leido.len()
);
assert_eq!(
leido, esperado,
"lo entregado no es lo escrito, o no está en el mismo orden"
);
}
#[test]
fn el_trozo_parcial_no_se_pierde() {
let TunelDelVolcado {
origen, productor, ..
} = tunel_de_prueba("volcado-desigual.sql", 2);
let trozos = [vec![b'a'; 97], vec![b'b'; 131]];
let esperado: Vec<u8> = trozos.concat();
let total = esperado.len();
let emisor = std::thread::spawn(move || {
for trozo in trozos {
productor
.enviar(trozo)
.expect("el origen sigue vivo, así que el canal no puede estar cerrado");
}
productor.terminar();
});
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let mut leido = Vec::new();
let mut bufer = [0_u8; 7];
loop {
let entregados = lector.read(&mut bufer).expect("leer no puede fallar");
if entregados == 0 {
break;
}
assert!(
entregados <= bufer.len(),
"una lectura entregó {entregados} bytes en un búfer de {}",
bufer.len()
);
leido.extend_from_slice(&bufer[..entregados]);
}
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
leido.len(),
total,
"se leyeron {} bytes de {total}: el trozo que sobraba de algún mensaje se perdió",
leido.len()
);
assert_eq!(
leido, esperado,
"lo leído no coincide con lo escrito: algún trozo parcial se perdió o se desordenó"
);
}
#[test]
fn el_enviar_bloquea_cuando_el_canal_esta_lleno_y_no_pierde_nada() {
let capacidad = 2;
let TunelDelVolcado {
origen,
productor,
entrega,
avance: _,
} = tunel_de_prueba("volcado-acotado.sql", capacidad);
let trozos: Vec<Vec<u8>> = (0..8).map(|i| vec![i as u8; 4096]).collect();
let esperado: Vec<u8> = trozos.concat();
let total = esperado.len();
let entregados = Arc::new(AtomicUsize::new(0));
let cuenta_del_hilo = Arc::clone(&entregados);
let emisor = std::thread::spawn(move || {
for trozo in trozos {
productor
.enviar(trozo)
.expect("el origen sigue vivo, así que el canal no puede estar cerrado");
cuenta_del_hilo.fetch_add(1, Ordering::SeqCst);
}
productor.terminar();
});
let limite = Instant::now() + Duration::from_secs(10);
while entregados.load(Ordering::SeqCst) < capacidad {
assert!(
!emisor.is_finished(),
"el productor terminó con menos de {capacidad} trozos entregados: el canal no \
está frenando al productor"
);
assert!(
Instant::now() < limite,
"el canal no se llenó en 10 segundos: la prueba no puede seguir"
);
std::thread::yield_now();
}
std::thread::sleep(Duration::from_millis(200));
assert_eq!(
entregados.load(Ordering::SeqCst),
capacidad,
"el productor entregó {} trozos con el canal lleno (capacidad {capacidad}): el canal \
no acota nada y el volcado se acumula en memoria",
entregados.load(Ordering::SeqCst)
);
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let mut leido = Vec::new();
let mut bufer = [0_u8; 7];
loop {
let entregados = lector
.read(&mut bufer)
.expect("con el productor avanzando, leer no puede fallar");
if entregados == 0 {
break;
}
leido.extend_from_slice(&bufer[..entregados]);
}
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
leido.len(),
total,
"se leyeron {} bytes de {total}: con el canal lleno se perdió lo que el productor \
había entregado",
leido.len()
);
assert_eq!(
leido, esperado,
"lo leído no coincide con lo escrito: el productor perdió o reordenó trozos con el \
canal lleno"
);
assert_eq!(
entregados.load(Ordering::SeqCst),
8,
"el productor no entregó los ocho trozos: alguno se quedó por el camino"
);
assert!(
entrega.completa(),
"el volcado se entregó entero y la señal sigue en falso"
);
}
#[test]
fn un_productor_que_termina_bien_deja_leer_todo_y_la_entrega_queda_completa() {
let TunelDelVolcado {
origen,
productor,
entrega,
avance: _,
} = tunel_de_prueba("volcado-entero.sql", 4);
assert!(
!entrega.completa(),
"un túnel recién hecho ya venía con la entrega completa"
);
let esperado = b"CREATE TABLE tienda (id INT);\nINSERT INTO tienda VALUES (1);\n".to_vec();
let a_enviar = esperado.clone();
let emisor = std::thread::spawn(move || {
productor
.enviar(a_enviar)
.expect("el canal no puede estar cerrado antes del final");
productor.terminar();
});
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let mut leido = Vec::new();
lector
.read_to_end(&mut leido)
.expect("con el final declarado, leer no puede fallar");
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
leido, esperado,
"lo leído no es el volcado que se envió, así que la señal hablaría de otra cosa"
);
assert!(
entrega.completa(),
"el productor declaró el final y el lector consumió el cierre, pero la señal sigue \
en falso: el motor daría por truncada una copia que llegó entera"
);
}
#[test]
fn un_productor_que_muere_sin_declarar_el_final_da_error_y_no_un_fin_de_fichero() {
let TunelDelVolcado {
origen,
productor,
entrega,
avance: _,
} = tunel_de_prueba("volcado-truncado.sql", 4);
productor
.enviar(b"parcial".to_vec())
.expect("el canal tiene sitio de sobra");
drop(productor);
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let mut bufer = [0_u8; 32];
let primeros = lector
.read(&mut bufer)
.expect("los bytes que sí llegaron se entregan igual");
assert_eq!(
&bufer[..primeros],
b"parcial",
"el trozo que sí llegó no se entregó"
);
let error = lector.read(&mut bufer).expect_err(
"un productor que muere sin declarar el final tiene que ser un error: devolver \
`Ok(0)` aquí archiva un volcado truncado como si fuera válido",
);
let motivo = error.to_string();
assert_eq!(
error.kind(),
std::io::ErrorKind::UnexpectedEof,
"el fallo de un cierre sin final declarado tiene que clasificarse como un final \
inesperado, no como otra cosa"
);
assert!(
motivo.contains("final"),
"el motivo no dice que faltó el final: {motivo}"
);
assert!(
motivo.contains("volcado-truncado.sql"),
"el motivo no nombra el volcado del que se cayó el productor: {motivo}"
);
assert!(
lector.read(&mut bufer).is_err(),
"una lectura posterior volvió a darse por buena tras un cierre sin final"
);
assert!(
!entrega.completa(),
"un productor que se cayó sin declarar el final dejó la señal de la entrega puesta: \
el motor daría por buena una copia que no llegó entera"
);
}
#[test]
fn el_canal_se_cierra_al_soltar_el_extremo_del_productor() {
let TunelDelVolcado {
origen,
productor,
entrega,
avance: _,
} = tunel_con_plazo("volcado-sin-final.sql", 4, Duration::from_secs(3600));
drop(productor);
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let (aviso, respuesta) = std::sync::mpsc::channel();
let hilo = std::thread::spawn(move || {
let mut bufer = [0_u8; 32];
let devuelto = lector.read(&mut bufer);
aviso
.send(devuelto)
.expect("la prueba sigue esperando la respuesta de la lectura");
});
let devuelto = respuesta.recv_timeout(Duration::from_secs(10)).expect(
"la lectura no volvió: el canal cerrado no la despertó, así que esperaba al plazo de \
una hora o a un emisor que ya no existe",
);
hilo.join().expect("el hilo de la lectura no puede caerse");
let error = devuelto.expect_err(
"un canal cerrado sin declarar el final es un volcado truncado: la lectura tiene que \
fallar, y nunca devolver 0",
);
assert_eq!(
error.kind(),
std::io::ErrorKind::UnexpectedEof,
"el desenlace de un canal cerrado sin final tiene que ser el del volcado truncado, no \
el del estancamiento: {error}"
);
assert!(
!entrega.completa(),
"un canal que se cerró sin declarar el final dejó la entrega marcada como completa"
);
}
#[test]
fn un_productor_callado_mas_que_el_plazo_da_error_de_estancamiento() {
let plazo = Duration::from_millis(200);
let TunelDelVolcado {
origen,
productor,
entrega,
avance: _,
} = tunel_con_plazo("volcado-atascado.sql", 4, plazo);
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let (aviso, respuesta) = std::sync::mpsc::channel();
let hilo = std::thread::spawn(move || {
let mut bufer = [0_u8; 32];
let devuelto = lector.read(&mut bufer);
aviso
.send(devuelto)
.expect("la prueba sigue esperando la respuesta de la lectura");
});
let devuelto = respuesta.recv_timeout(Duration::from_secs(30)).expect(
"la lectura no volvió por el plazo: se quedó esperando a un productor callado con el \
extremo de envío vivo, que es el cuelgue que este plazo existe para cortar",
);
hilo.join().expect("el hilo de la lectura no puede caerse");
let error = devuelto.expect_err(
"un productor callado más que el plazo tiene que dar error, y nunca un fin de fichero",
);
assert_eq!(
error.kind(),
std::io::ErrorKind::TimedOut,
"el desenlace de una espera agotada tiene que clasificarse como un tiempo agotado, no \
como otra cosa: {error}"
);
let motivo = error.to_string();
assert!(
motivo.contains("volcado-atascado.sql"),
"el motivo del estancamiento no nombra el volcado: {motivo}"
);
assert!(
motivo.contains(&format!("{plazo:?}")),
"el motivo del estancamiento no nombra el plazo que se agotó ({plazo:?}): {motivo}"
);
assert!(
!entrega.completa(),
"una lectura estancada dejó la entrega marcada como completa"
);
drop(productor);
}
#[test]
fn los_entries_no_se_pueden_pedir_dos_veces() {
let TunelDelVolcado {
origen, productor, ..
} = tunel_de_prueba("volcado-de-prueba.sql", 4);
let primera: Vec<_> = origen.entries().collect();
assert_eq!(
primera.len(),
1,
"la primera llamada tiene que dar una entrada"
);
assert!(
primera[0].is_ok(),
"la primera entrada no puede ser un error"
);
let segunda: Vec<_> = origen.entries().collect();
assert!(
!segunda.is_empty(),
"la segunda llamada dio un iterador vacío: el motor que pidió las entradas dos veces \
no tendría forma de enterarse de su error"
);
let error = segunda[0]
.as_ref()
.err()
.expect("la segunda llamada tiene que dar un error, no una entrada");
assert!(
error.to_string().contains("volcado-de-prueba.sql"),
"el motivo del segundo uso no nombra el volcado: {error}"
);
drop(productor);
}
#[test]
fn los_entries_dan_una_entrada_con_su_nombre_y_su_nodo_de_fichero() {
let TunelDelVolcado {
origen, productor, ..
} = tunel_de_prueba("volcado-de-tienda.sql", 4);
let mut entradas = origen.entries();
let entrada = entradas
.next()
.expect("el origen tiene que dar una entrada")
.expect("la entrada no puede ser un error");
assert!(
entradas.next().is_none(),
"un volcado es una sola entrada, y salieron más"
);
assert_eq!(
entrada.path,
PathBuf::from("volcado-de-tienda.sql"),
"el camino de la entrada no es el nombre con el que se pidió el volcado"
);
assert_eq!(
entrada.node.node_type,
NodeType::File,
"el nodo del volcado tiene que ser un fichero: lo que entra por aquí son bytes"
);
assert!(
entrada.open.is_some(),
"la entrada no trae cómo abrirse, así que la biblioteca no puede leer nada"
);
drop(productor);
}
#[test]
fn la_entrega_no_esta_completa_mientras_el_volcado_va_a_medias() {
let tunel = tunel_de_prueba("volcado-a-medias.sql", 4);
assert!(
!tunel.entrega.completa(),
"un túnel recién hecho ya venía con la entrega completa"
);
let primer_trozo = b"CREATE TABLE tienda (id INT);\n".to_vec();
tunel
.productor
.enviar(primer_trozo.clone())
.expect("el canal tiene sitio de sobra");
let mut lector = lector_de(&tunel.origen).expect("el lector tiene que abrirse");
let mut bufer = [0_u8; 128];
let entregados = lector
.read(&mut bufer)
.expect("los bytes que sí llegaron se entregan igual");
assert_eq!(
&bufer[..entregados],
primer_trozo.as_slice(),
"el primer trozo del volcado no se entregó"
);
assert!(
!tunel.entrega.completa(),
"el volcado estaba a medias y la señal de la entrega ya estaba puesta: el motor daría \
por buena una copia que no llegó entera"
);
}
#[test]
fn el_extremo_del_productor_no_se_puede_clonar() {
assert!(
SondaDeClon::<String>::nueva().es_clonable(),
"el sondeo no reconoce un tipo que sí es `Clone`, así que esta prueba estaría pasando \
sin comprobar nada"
);
assert!(
!SondaDeClon::<ProductorDelVolcado>::nueva().es_clonable(),
"`ProductorDelVolcado` se puede clonar: con dos extremos vivos el canal no se cierra \
cuando el productor termina, y el lector espera al productor que ya no está"
);
}
struct SondaDeClon<T>(PhantomData<T>);
impl<T> SondaDeClon<T> {
const fn nueva() -> Self {
SondaDeClon(PhantomData)
}
}
trait EsClonable {
fn es_clonable(&self) -> bool {
false
}
}
impl<T> EsClonable for SondaDeClon<T> {}
impl<T: Clone> SondaDeClon<T> {
fn es_clonable(&self) -> bool {
true
}
}
#[test]
fn una_lectura_con_bufer_vacio_no_espera_ni_consume_nada() {
let TunelDelVolcado {
origen,
productor,
entrega,
avance: _,
} = tunel_con_plazo("volcado-sin-bufer.sql", 4, Duration::from_secs(60));
let lector = lector_de(&origen).expect("el lector tiene que abrirse");
let (aviso, respuesta) = std::sync::mpsc::channel();
let hilo_vacio = std::thread::spawn(move || {
let mut lector = lector;
let mut nada: [u8; 0] = [];
let devuelto = lector.read(&mut nada);
aviso
.send(devuelto)
.expect("la prueba sigue esperando la respuesta de la lectura vacía");
lector
});
let devuelto = respuesta.recv_timeout(Duration::from_secs(5)).expect(
"la lectura con el búfer vacío se quedó esperando un mensaje que nadie había \
mandado: el contrato de `Read` obliga a contestar `0` en el acto",
);
assert_eq!(
devuelto.expect("un búfer vacío no puede fallar"),
0,
"el contrato de `Read` obliga a contestar `0` con un búfer vacío"
);
assert!(
!entrega.completa(),
"la lectura con el búfer vacío se tomó por un final de fichero del volcado"
);
let mut lector = hilo_vacio
.join()
.expect("el hilo de la lectura vacía no puede caerse");
let esperado = b"CREATE TABLE tienda (id INT);\n".to_vec();
let a_enviar = esperado.clone();
let emisor = std::thread::spawn(move || {
productor
.enviar(a_enviar)
.expect("el canal no puede estar cerrado antes del final");
productor.terminar();
});
let mut bufer = [0_u8; 5];
let entregados = lector
.read(&mut bufer)
.expect("con el volcado en marcha, leer no puede fallar");
assert_eq!(
&bufer[..entregados],
&esperado[..entregados],
"la primera lectura de verdad no recibió el principio del volcado: la lectura vacía \
se lo había comido"
);
let mut resto = Vec::new();
lector
.read_to_end(&mut resto)
.expect("con el final declarado, leer no puede fallar");
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
[bufer[..entregados].as_ref(), resto.as_slice()].concat(),
esperado,
"entre la lectura vacía y las siguientes no salió el volcado entero"
);
assert!(
entrega.completa(),
"el volcado se entregó entero y la señal sigue en falso"
);
}
#[test]
fn un_nombre_de_volcado_vacio_se_rechaza() {
assert_eq!(
nombre_rechazado(tunel_del_volcado("", 4, PLAZO_DE_PRUEBA)),
"",
"el error no señala el nombre vacío como el que rechazó"
);
}
#[test]
fn un_nombre_de_volcado_de_punto_se_rechaza() {
assert_eq!(
nombre_rechazado(tunel_del_volcado(".", 4, PLAZO_DE_PRUEBA)),
".",
"el error no señala el punto como el nombre que rechazó"
);
}
#[test]
fn un_nombre_de_volcado_de_punto_punto_se_rechaza() {
assert_eq!(
nombre_rechazado(tunel_del_volcado("..", 4, PLAZO_DE_PRUEBA)),
"..",
"el error no señala los dos puntos como el nombre que rechazó"
);
}
#[test]
fn un_nombre_de_volcado_con_barra_se_rechaza() {
assert_eq!(
nombre_rechazado(tunel_del_volcado("volcados/tienda.sql", 4, PLAZO_DE_PRUEBA)),
"volcados/tienda.sql",
"el error no señala el nombre con barra como el que rechazó"
);
}
#[test]
fn un_nombre_de_volcado_con_barra_invertida_se_rechaza() {
assert_eq!(
nombre_rechazado(tunel_del_volcado(
"volcados\\tienda.sql",
4,
PLAZO_DE_PRUEBA
)),
"volcados\\tienda.sql",
"el error no señala el nombre con barra invertida como el que rechazó"
);
}
#[test]
fn un_nombre_de_volcado_normal_si_pasa() {
let tunel = tunel_de_prueba("volcado-de-tienda-2026-10-25.sql", 4);
assert!(
!tunel.entrega.completa(),
"un túnel recién construido no puede traer la entrega completa"
);
assert_eq!(
tunel.origen.nombre, "volcado-de-tienda-2026-10-25.sql",
"el nombre aceptado no es el que el origen va a usar como camino de la entrada"
);
}
#[test]
fn el_avance_empieza_en_cero_y_crece_con_lo_entregado() {
let TunelDelVolcado {
origen: _origen,
productor,
entrega: _entrega,
avance,
} = tunel_de_prueba("volcado-contado.sql", 8);
assert_eq!(
avance.bytes(),
0,
"un túnel recién construido no puede traer bytes contados"
);
let trozos = [vec![b'a'; 97], vec![b'b'; 1024], vec![b'c'; 3]];
let total: u64 = trozos.iter().map(|trozo| trozo.len() as u64).sum();
for trozo in trozos {
productor
.enviar(trozo)
.expect("el canal tiene sitio de sobra, así que el envío no puede fallar");
}
assert_eq!(
avance.bytes(),
total,
"la cuenta no es la suma de lo entregado: {} de {total}",
avance.bytes()
);
drop(productor);
}
#[test]
fn un_trozo_vacio_no_cuenta_ni_viaja_por_el_canal() {
let TunelDelVolcado {
origen,
productor,
entrega: _entrega,
avance,
} = tunel_de_prueba("volcado-sin-trozo-vacio.sql", 4);
productor
.enviar(Vec::new())
.expect("un trozo vacío no puede fallar: ni siquiera se envía");
assert_eq!(
avance.bytes(),
0,
"un trozo vacío sumó a la cuenta, que es contar lo que no se entregó"
);
let esperado = b"CREATE TABLE tienda (id INT);\n".to_vec();
let a_enviar = esperado.clone();
let emisor = std::thread::spawn(move || {
productor
.enviar(a_enviar)
.expect("el canal no puede estar cerrado antes del final");
productor.terminar();
});
let mut lector = lector_de(&origen).expect("el lector tiene que abrirse");
let mut leido = Vec::new();
lector
.read_to_end(&mut leido)
.expect("con el final declarado, leer no puede fallar");
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
leido, esperado,
"lo que llegó al lector no es el trozo de verdad: el vacío viajó por el canal"
);
assert_eq!(
avance.bytes(),
esperado.len() as u64,
"la cuenta no cuadra con lo que sí se entregó"
);
}
#[test]
fn el_avance_se_puede_leer_mientras_el_productor_empuja_y_no_retrocede() {
const TROZOS: usize = 8;
const BYTES: usize = 100;
let TunelDelVolcado {
origen: _origen,
productor,
entrega: _entrega,
avance,
} = tunel_de_prueba("volcado-leido-en-vivo.sql", TROZOS + 1);
let total = (TROZOS * BYTES) as u64;
let emisor = std::thread::spawn(move || {
for _ in 0..TROZOS {
productor
.enviar(vec![b'x'; BYTES])
.expect("el canal tiene sitio de sobra, así que el envío no puede fallar");
}
productor.terminar();
});
let limite = Instant::now() + Duration::from_secs(10);
let mut anterior = 0;
let mut muestras = 0;
while avance.bytes() < total {
let ahora = avance.bytes();
assert!(
ahora >= anterior,
"la cuenta retrocedió, y un avance que baja es un panel que va hacia atrás: \
{anterior} -> {ahora}"
);
anterior = ahora;
muestras += 1;
assert!(
Instant::now() < limite,
"la cuenta no llegó al total en 10 segundos: {ahora} de {total}"
);
std::thread::yield_now();
}
emisor.join().expect("el hilo productor no puede caerse");
assert_eq!(
avance.bytes(),
total,
"la cuenta final no es la de lo entregado"
);
assert!(
muestras > 0,
"el bucle no llegó a leer la cuenta: la prueba no midió nada"
);
assert!(
anterior <= total,
"alguna muestra pasó del total: {anterior} de {total}"
);
}
#[test]
fn el_debug_del_avance_publica_la_cuenta_y_nada_del_volcado() {
let distinguible = "contenido-distinguible-del-volcado-0002";
let tunel = tunel_de_prueba("volcado-sin-publicar.sql", 4);
tunel
.productor
.enviar(distinguible.as_bytes().to_vec())
.expect("el canal tiene sitio de sobra");
let en_debug = format!("{:?}", tunel.avance);
assert!(
en_debug.contains("AvanceDelVolcado"),
"el Debug no dice ni de qué tipo es: {en_debug}"
);
assert!(
en_debug.contains(&tunel.avance.bytes().to_string()),
"el Debug no publica la cuenta, que es lo único que tiene que publicar: {en_debug}"
);
assert!(
!en_debug.contains(distinguible),
"el Debug de la cuenta publicó el contenido del volcado: {en_debug}"
);
assert!(
!en_debug.contains("volcado-sin-publicar.sql"),
"el Debug de la cuenta publicó el nombre del volcado: {en_debug}"
);
drop(tunel.productor);
}
}