liminal-sdk 0.19.2

use super::*;
use std::net::TcpListener;
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))
}

fn response() -> Result<(Frame, Vec<u8>), Box<dyn std::error::Error>> {
    let frame = Frame::ConnectAck {
        flags: 0,
        selected_version: CLIENT_MAX_VERSION,
        capabilities: 0,
    };
    let mut bytes = vec![0; encoded_len(&frame)?];
    let count = encode(&frame, &mut bytes)?;
    bytes.truncate(count);
    Ok((frame, bytes))
}

#[test]
fn idle_read_wake_retains_a_partial_frame_and_the_live_socket() -> TestResult {
    let (mut connection, mut peer, wake) = connected()?;
    let (frame, bytes) = response()?;
    let split = bytes.len() / 2;
    connection.buffer.extend_from_slice(&bytes[..split]);
    wake.wake()?;
    wake.wake()?;
    assert!(matches!(
        connection.receive_optional_within(IO_TIMEOUT),
        Err(SdkError::ReadInterrupted)
    ));
    assert_eq!(connection.buffer, bytes[..split]);
    peer.write_all(&bytes[split..])?;
    assert_eq!(connection.receive()?, frame);
    assert!(connection.buffer.is_empty());
    Ok(())
}

#[test]
fn idle_read_wake_does_not_interrupt_a_reply_owed_read() -> TestResult {
    let (mut connection, mut peer, wake) = connected()?;
    let (frame, bytes) = response()?;
    wake.wake()?;
    peer.write_all(&bytes)?;
    assert_eq!(connection.receive()?, frame);
    // The wake remains pending for its intended idle receive.
    assert!(matches!(
        connection.receive_optional_within(IO_TIMEOUT),
        Err(SdkError::ReadInterrupted)
    ));
    Ok(())
}

/// The reader proves it reached the blocking `poll` with no wake pending
/// before the wake is sent, so the interrupt can only arrive through the wait
/// itself. The old shape signalled before the receive was even entered and
/// passed just as well against a wake that was already pending.
#[test]
fn idle_read_wake_interrupts_a_reader_already_blocked_in_its_wait() -> TestResult {
    let (mut connection, peer, wake) = connected()?;
    let (entered, entering) = mpsc::channel();
    wake.probe_waits(move |step, _| {
        if let super::super::read_wake::WaitStep::Polling { pending } = step {
            let _sent = entered.send(pending);
        }
    });
    let (finished, completion) = mpsc::channel();
    let worker = std::thread::spawn(move || {
        let result = connection.receive_optional_within(IO_TIMEOUT);
        let _sent = finished.send(result);
    });
    let pending_on_entry = entering.recv_timeout(Duration::from_secs(1))?;
    assert!(
        !pending_on_entry,
        "the reader entered its wait with a wake already pending"
    );
    wake.wake()?;
    let result = completion.recv_timeout(Duration::from_secs(1))?;
    worker.join().map_err(|_| "idle reader panicked")?;
    assert!(matches!(result, Err(SdkError::ReadInterrupted)));
    assert!(
        entering.try_recv().is_err(),
        "the wake must end the wait the reader was blocked in, not a later one"
    );
    drop(peer);
    Ok(())
}

/// A wake sent while the reader is claiming an earlier one is a new request,
/// and the next idle receive must see it. Clearing the flag only after the
/// pipe byte is taken absorbs it: the second wake finds the flag still set,
/// writes nothing, and the next receive waits out its whole budget.
#[test]
fn a_wake_sent_while_an_earlier_one_is_claimed_reaches_the_next_receive() -> TestResult {
    let (mut connection, peer, wake) = connected()?;
    let (second_sent, second_result) = mpsc::channel();
    let mut fired = false;
    wake.probe_waits(move |step, wake| {
        if matches!(step, super::super::read_wake::WaitStep::Claiming) && !fired {
            fired = true;
            let _sent = second_sent.send(wake.wake());
        }
    });
    wake.wake()?;
    assert!(matches!(
        connection.receive_optional_within(IO_TIMEOUT),
        Err(SdkError::ReadInterrupted)
    ));
    second_result.recv_timeout(Duration::from_secs(1))??;
    let started = Instant::now();
    let second = connection.receive_optional_within(IO_TIMEOUT);
    assert!(
        matches!(second, Err(SdkError::ReadInterrupted)),
        "the second wake was absorbed: the next receive returned {second:?} after {:?}",
        started.elapsed()
    );
    drop(peer);
    Ok(())
}