liminal-sdk 0.19.6

//! SDK-023 R1: a write never waits for a read on the same connection.

use super::*;
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::mpsc;
use std::time::Duration as StdDuration;

use liminal::protocol::{PUBLISH_ACK_POSITION_VERSION, decode, encode, encoded_len};
use liminal_protocol::wire::{ClientRequest, EnrollmentRequest, EnrollmentToken};

use super::super::read_wake::WaitStep;

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

/// How long the test lets the server wait for the second thread's frame
/// before it ends the parked read itself; a guard on the test, never a wait
/// the transport makes.
const GUARD: StdDuration = StdDuration::from_secs(2);

/// Accepts one connection, answers its `Connect` with a `ConnectAck`, then
/// tells `seen` the moment any further byte arrives, and keeps the socket
/// open until `hold` is dropped.
fn fake_server(
    listener: TcpListener,
    seen: mpsc::Sender<()>,
    hold: mpsc::Receiver<()>,
) -> std::thread::JoinHandle<Result<(), String>> {
    std::thread::spawn(move || {
        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 let Ok((_, used)) = decode(&buffer) {
                buffer.drain(..used);
                break;
            }
        }
        let ack = Frame::ConnectAck {
            flags: 0,
            selected_version: PUBLISH_ACK_POSITION_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())?;
        socket
            .write_all(&bytes[..count])
            .map_err(|error| error.to_string())?;
        if buffer.is_empty() {
            let read = socket.read(&mut chunk).map_err(|error| error.to_string())?;
            if read == 0 {
                return Ok(());
            }
        }
        let _sent = seen.send(());
        let _held = hold.recv();
        Ok(())
    })
}

#[test]
fn a_send_reaches_the_server_while_another_thread_is_parked_in_a_read() -> TestResult {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = ServerAddress::new(listener.local_addr()?.to_string())?;
    let (seen, saw) = mpsc::channel();
    let (release, hold) = mpsc::channel::<()>();
    let server = fake_server(listener, seen, hold);
    let transport = Arc::new(TcpRemoteTransport::connect(&address)?);

    let (parked, parking) = mpsc::channel();
    transport.read_wake.probe_waits(move |step, _| {
        if matches!(step, WaitStep::Polling { .. }) {
            let _sent = parked.send(());
        }
    });
    let reader = {
        let transport = Arc::clone(&transport);
        let address = address.clone();
        std::thread::spawn(move || {
            transport
                .receive_participant_within(&address, Duration::MAX)
                .map(|frame| frame.is_some())
        })
    };
    parking.recv_timeout(GUARD)?;

    let writer = {
        let transport = Arc::clone(&transport);
        std::thread::spawn(move || {
            transport.send_participant(
                &address,
                &ClientRequest::Enrollment(EnrollmentRequest {
                    conversation_id: 1,
                    enrollment_token: EnrollmentToken::new([3; 16]),
                }),
            )
        })
    };
    let reached_while_parked = saw.recv_timeout(GUARD).is_ok();
    let reader_still_parked = !reader.is_finished();

    // End the parked read by its own wake, so the test ends either way.
    transport.read_wake.wake()?;
    let read = reader.join().map_err(|_| "the reader panicked")?;
    writer.join().map_err(|_| "the writer panicked")??;
    drop(release);
    server.join().map_err(|_| "the server panicked")??;

    assert!(
        reached_while_parked && reader_still_parked,
        "the second thread's send must reach the server while the first is parked in its read \
         (reached {reached_while_parked}, reader still parked {reader_still_parked})"
    );
    assert!(
        matches!(read, Err(SdkError::ReadInterrupted)),
        "the parked read must end on its wake, got {read:?}"
    );
    Ok(())
}