liminal-sdk 0.20.20

//! SDK-026 R1: a TCP read keeps the answer it read, whatever the window does.
//!
//! XNU refuses every socket option with `EINVAL` once both directions of the
//! socket are shut, which is what a peer's reset leaves. A server that answers
//! and then closes can land its close before the client touches the window,
//! so the window may be refused after the answer is read (a restore) or
//! before the read begins (an arm). Neither may cost the client the answer.

use super::*;

/// `EINVAL` on macOS and Linux alike.
const EINVAL: i32 = 22;

/// A [`FrameStream`] that hands over `ready` and refuses every window
/// operation with `EINVAL` from a chosen point, as XNU does once the peer has
/// reset the link. Writes are swallowed. An empty `ready` reads end of file.
struct RefusingStream {
    /// Bytes handed over, drained front-first.
    ready: Vec<u8>,
    /// Every window operation is refused while set.
    refusing: bool,
    /// Set `refusing` once `ready` has been handed over: the reset landing
    /// after the answer.
    refuse_once_drained: bool,
}

impl RefusingStream {
    /// Refuses the window only once `ready` has been read.
    const fn after_the_answer(ready: Vec<u8>) -> Self {
        Self {
            ready,
            refusing: false,
            refuse_once_drained: true,
        }
    }

    /// Refuses the window from the start, with `ready` already buffered.
    const fn from_the_start(ready: Vec<u8>) -> Self {
        Self {
            ready,
            refusing: true,
            refuse_once_drained: false,
        }
    }

    fn window(&self) -> io::Result<()> {
        if self.refusing {
            return Err(io::Error::from_raw_os_error(EINVAL));
        }
        Ok(())
    }
}

impl FrameStream for RefusingStream {
    fn read_bytes(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let take = self.ready.len().min(buf.len());
        let Some(target) = buf.get_mut(..take) else {
            return Ok(0);
        };
        target.copy_from_slice(self.ready.drain(..take).as_slice());
        if self.ready.is_empty() && self.refuse_once_drained {
            self.refusing = true;
        }
        Ok(take)
    }

    fn write_all_bytes(&mut self, _bytes: &[u8]) -> io::Result<()> {
        Ok(())
    }

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

    fn set_read_deadline(&mut self, _timeout: Duration) -> io::Result<()> {
        self.window()
    }

    fn clear_read_deadline(&mut self) -> io::Result<()> {
        self.window()
    }
}

/// A connection over `stream`, past its open.
fn over(stream: RefusingStream) -> Connection<RefusingStream> {
    Connection {
        #[cfg(unix)]
        read_wake: None,
        stream,
        buffer: Vec::new(),
        open_conversations: BTreeSet::new(),
        pump_fences_pending: 0,
        pump_diagnostics: ParticipantPumpDiagnostics::default(),
        negotiated_version: None,
    }
}

/// The exact wire bytes of `frame`.
fn bytes_of(frame: &Frame) -> Result<Vec<u8>, SdkError> {
    let len = encoded_len(frame).map_err(|error| protocol_error(&error))?;
    let mut bytes = vec![0_u8; len];
    let written = encode(frame, &mut bytes).map_err(|error| protocol_error(&error))?;
    bytes.truncate(written);
    Ok(bytes)
}

/// The answer every test reads.
const fn answer() -> Frame {
    Frame::ConnectAck {
        flags: 0,
        selected_version: CLIENT_MAX_VERSION,
        capabilities: 0,
    }
}

#[test]
fn a_reply_owed_read_keeps_its_answer_when_the_window_is_refused_after_it() -> Result<(), SdkError>
{
    let mut connection = over(RefusingStream::after_the_answer(bytes_of(&answer())?));
    let replied = connection.round_trip(&Frame::Ping { flags: 0 })?;
    assert_eq!(replied, answer(), "the answer read before the reset");
    Ok(())
}

#[test]
fn a_bounded_read_keeps_its_answer_when_the_window_is_refused_after_it() -> Result<(), SdkError> {
    let mut connection = over(RefusingStream::after_the_answer(bytes_of(&answer())?));
    let replied = connection.receive_optional_within(Duration::from_secs(30))?;
    assert_eq!(replied, Some(answer()), "the answer read before the reset");
    Ok(())
}

#[test]
fn a_conversation_drain_keeps_its_answer_when_the_window_is_refused_after_it()
-> Result<(), SdkError> {
    let mut connection = over(RefusingStream::after_the_answer(bytes_of(&answer())?));
    let replied = connection.receive_with_timeout(CONVERSATION_DRAIN_TIMEOUT)?;
    assert_eq!(replied, Some(answer()), "the answer read before the reset");
    Ok(())
}

#[test]
fn a_pump_fence_keeps_its_quiet_answer_when_the_window_is_refused_after_it() -> Result<(), SdkError>
{
    let mut connection = over(RefusingStream::after_the_answer(bytes_of(&Frame::Pong {
        flags: 0,
    })?));
    let drained = connection.try_receive_participant()?;
    assert!(
        drained.is_none(),
        "the fence's Pong was read before the reset, so the drain is quiet: {drained:?}"
    );
    Ok(())
}

#[test]
fn a_reply_owed_read_keeps_a_buffered_answer_when_the_window_is_refused_before_it()
-> Result<(), SdkError> {
    let mut connection = over(RefusingStream::from_the_start(bytes_of(&answer())?));
    let replied = connection.round_trip(&Frame::Ping { flags: 0 })?;
    assert_eq!(replied, answer(), "the answer buffered before the reset");
    Ok(())
}

#[test]
fn a_read_that_finds_the_end_names_it_and_the_refused_window() -> Result<(), SdkError> {
    let mut connection = over(RefusingStream::from_the_start(Vec::new()));
    let outcome = connection.round_trip(&Frame::Ping { flags: 0 });
    let Err(SdkError::Connection { description }) = outcome else {
        return Err(SdkError::Protocol {
            description: format!(
                "a peer that is gone must fail the read as a lost connection, got {outcome:?}"
            ),
        });
    };
    assert!(
        description.contains("closed the connection"),
        "the read's own end of file is named: {description}"
    );
    assert!(
        description.contains("os error 22"),
        "the refused window is named beside it: {description}"
    );
    Ok(())
}