liminal-server 0.14.3

//! An abruptly dropped peer is a transport loss, not a server process crash.

use std::io::Write;
use std::sync::{Arc, Mutex, PoisonError};

use beamr::process::ExitReason;

use super::SliceStep;
use super::tests::scheduler_free_process;

#[derive(Clone)]
struct CaptureWriter(Arc<Mutex<Vec<u8>>>);

impl Write for CaptureWriter {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.0
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

fn capture<T>(body: impl FnOnce() -> T) -> (T, String) {
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let writer = CaptureWriter(Arc::clone(&buffer));
    let subscriber = tracing_subscriber::fmt()
        .with_writer(move || writer.clone())
        .with_ansi(false)
        .with_max_level(tracing::Level::DEBUG)
        .finish();
    let result = tracing::subscriber::with_default(subscriber, body);
    let bytes = buffer.lock().unwrap_or_else(PoisonError::into_inner);
    (result, String::from_utf8_lossy(&bytes).into_owned())
}

#[test]
fn peer_drop_without_close_handshake_is_not_a_process_crash() -> Result<(), String> {
    let (mut process, client) = scheduler_free_process()?;
    drop(client);

    let (step, log) = capture(|| process.service_socket(1));

    assert!(matches!(step, SliceStep::Stop(ExitReason::Error)));
    assert!(log.contains("websocket peer reset"), "{log}");
    assert!(log.contains("without closing handshake"), "{log}");
    assert!(!log.contains("connection process crashed"), "{log}");
    Ok(())
}

#[test]
fn missing_server_socket_still_reports_a_process_crash() -> Result<(), String> {
    let (mut process, _client) = scheduler_free_process()?;
    process.socket = None;

    let (step, log) = capture(|| process.service_socket(1));

    assert!(matches!(step, SliceStep::Stop(ExitReason::Error)));
    assert!(log.contains("connection process crashed"), "{log}");
    assert!(!log.contains("websocket peer reset"), "{log}");
    Ok(())
}

#[test]
fn socket_reset_is_distinguished_from_other_io_failures() -> Result<(), String> {
    for (kind, peer_reset) in [
        (std::io::ErrorKind::ConnectionReset, true),
        (std::io::ErrorKind::ConnectionAborted, true),
        (std::io::ErrorKind::BrokenPipe, true),
        (std::io::ErrorKind::UnexpectedEof, true),
        (std::io::ErrorKind::PermissionDenied, false),
    ] {
        let (mut process, client) = scheduler_free_process()?;
        drop(client);
        let error = tungstenite::Error::Io(std::io::Error::from(kind));
        let (step, log) = capture(|| process.finish_read_error(1, &error));

        assert!(matches!(step, SliceStep::Stop(ExitReason::Error)));
        assert_eq!(log.contains("websocket peer reset"), peer_reset, "{log}");
        assert_eq!(
            log.contains("connection process crashed"),
            !peer_reset,
            "{log}"
        );
    }
    Ok(())
}