liminal-sdk 0.19.7

//! SDK-023 R2 and R3: a read with no budget waits in one poll, and the
//! connected socket carries the server's keepalive.

use super::*;
use std::net::TcpListener;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn connected()
-> Result<(Connection<TcpStream>, TcpStream, super::super::ReadWake), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let stream = TcpStream::connect(listener.local_addr()?)?;
    let (peer, _) = listener.accept()?;
    let wake = super::super::ReadWake::new()?;
    let connection = Connection {
        stream,
        read_wake: Some(wake.clone()),
        buffer: Vec::new(),
        open_conversations: BTreeSet::new(),
        pump_fences_pending: 0,
        pump_diagnostics: ParticipantPumpDiagnostics::default(),
        negotiated_version: None,
    };
    Ok((connection, peer, wake))
}

/// Long enough that a 5 s re-arm would poll three times; the test's own
/// measuring span, not a wait the transport makes.
const PARKED: Duration = Duration::from_secs(11);

#[test]
fn a_parked_read_polls_once() -> TestResult {
    let (mut connection, peer, wake) = connected()?;
    let polls = Arc::new(AtomicUsize::new(0));
    let (entered, entering) = mpsc::channel();
    {
        let polls = Arc::clone(&polls);
        wake.probe_waits(move |step, _| {
            if matches!(step, super::super::read_wake::WaitStep::Polling { .. }) {
                polls.fetch_add(1, Ordering::AcqRel);
                let _sent = entered.send(());
            }
        });
    }
    let reader = std::thread::spawn(move || connection.receive_optional_within(Duration::MAX));
    entering.recv_timeout(Duration::from_secs(1))?;
    std::thread::sleep(PARKED);
    wake.wake()?;
    let result = reader.join().map_err(|_| "the parked reader panicked")?;
    drop(peer);
    assert!(matches!(result, Err(SdkError::ReadInterrupted)));
    assert_eq!(
        polls.load(Ordering::Acquire),
        1,
        "a read with no budget must wait in one poll for {PARKED:?}, not re-arm on a tick"
    );
    Ok(())
}

/// The keepalive the server sets on every socket it accepts
/// (liminal-server `accepted_socket.rs`), read back from a client socket.
fn assert_server_keepalive(stream: &TcpStream, which: &str) -> TestResult {
    let socket = socket2::SockRef::from(stream);
    assert!(socket.keepalive()?, "{which}: SO_KEEPALIVE must be on");
    assert_eq!(
        socket.tcp_keepalive_time()?,
        Duration::from_secs(30),
        "{which}: idle"
    );
    assert_eq!(
        socket.tcp_keepalive_interval()?,
        Duration::from_secs(10),
        "{which}: interval"
    );
    assert_eq!(socket.tcp_keepalive_retries()?, 5, "{which}: retries");
    Ok(())
}

#[test]
fn a_connected_socket_carries_the_servers_keepalive() -> TestResult {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?.to_string();
    let server = std::thread::spawn(move || -> Result<(), String> {
        let (mut socket, _) = listener.accept().map_err(|error| error.to_string())?;
        let mut buffer = Vec::new();
        let mut chunk = [0_u8; 4096];
        loop {
            let read = socket.read(&mut chunk).map_err(|error| error.to_string())?;
            if read == 0 {
                return Err("the client closed before its Connect".to_owned());
            }
            buffer.extend_from_slice(&chunk[..read]);
            if decode(&buffer).is_ok() {
                break;
            }
        }
        let ack = Frame::ConnectAck {
            flags: 0,
            selected_version: CLIENT_MAX_VERSION,
            capabilities: 0,
        };
        let mut bytes = vec![0; encoded_len(&ack).map_err(|error| error.to_string())?];
        let count = encode(&ack, &mut bytes).map_err(|error| error.to_string())?;
        std::io::Write::write_all(&mut socket, &bytes[..count])
            .map_err(|error| error.to_string())?;
        let _closed = socket.read(&mut chunk);
        Ok(())
    });
    let connection = Connection::connect_with_auth(&address, &[])?;
    let checked = assert_server_keepalive(&connection.stream, "the transport's socket");
    drop(connection);
    server.join().map_err(|_| "the server panicked")??;
    checked
}

#[test]
fn the_push_clients_socket_carries_the_servers_keepalive() -> TestResult {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?.to_string();
    let stream = super::super::tcp::connect_push_socket_for_tests(&address)?;
    let (_peer, _) = listener.accept()?;
    assert_server_keepalive(&stream, "the push client's socket")
}

/// Longer than one 5 s socket window: the peer's hold, the test's measuring
/// span, not a wait the transport makes.
const HELD_REPLY: Duration = Duration::from_secs(6);
/// Inside the hold, when the reader is parked on the socket.
const PARKED_CHECK: Duration = Duration::from_secs(3);

#[test]
fn a_reply_owed_read_waits_with_no_window() -> TestResult {
    let (mut connection, mut peer, _wake) = connected()?;
    // The steady-state window connect_with_auth leaves on the socket.
    connection.stream.set_read_timeout(Some(IO_TIMEOUT))?;
    let socket = connection.stream.try_clone()?;
    let reader = std::thread::spawn(move || connection.receive());
    std::thread::sleep(PARKED_CHECK);
    let window_while_parked = socket.read_timeout()?;
    std::thread::sleep(HELD_REPLY.saturating_sub(PARKED_CHECK));
    let pong = Frame::Pong { flags: 0 };
    let mut bytes = vec![0; encoded_len(&pong)?];
    let count = encode(&pong, &mut bytes)?;
    std::io::Write::write_all(&mut peer, &bytes[..count])?;
    let answered = reader
        .join()
        .map_err(|_| "the reply-owed reader panicked")??;
    assert!(matches!(answered, Frame::Pong { .. }));
    assert_eq!(
        window_while_parked, None,
        "a read that owes a reply must wait with no socket window beneath it"
    );
    assert_eq!(
        socket.read_timeout()?,
        Some(IO_TIMEOUT),
        "the steady-state window is restored once the reply is read"
    );
    Ok(())
}