diff --git a/README.md b/README.md index a5016bf8..4c6c4512 100644 --- a/README.md +++ b/README.md @@ -39,11 +39,10 @@ These current host-level limitations may affect application design: configurable size limit ([issue #436](https://github.com/roc-lang/basic-cli/issues/436)), and several network and TLS failures currently use a generic transport error ([issue #438](https://github.com/roc-lang/basic-cli/issues/438)). -- TCP connect, read, and write operations have no caller-configurable timeouts. - They can block until the operation completes or the operating system returns - an error. `Tcp.Stream.read_until!` and `read_line!` also have no size limit and - buffer until the delimiter or EOF. [Issue #437](https://github.com/roc-lang/basic-cli/issues/437) - tracks timeout and bounded-read APIs. +- TCP connect, read, and write operations require a caller-configured timeout + covering the whole operation. A zero timeout fails immediately. + `Tcp.Stream.read_until!` and `read_line!` also require a maximum byte count, + preventing an untrusted peer from causing unbounded buffering. - SQLite keeps up to 16 recently used ordinary path connections open. A live prepared statement keeps its connection open after cache eviction, and reused paths continue sharing that live connection. The exact `:memory:` path is kept diff --git a/examples/tcp-client.roc b/examples/tcp-client.roc index 24eb426c..2834e11d 100644 --- a/examples/tcp-client.roc +++ b/examples/tcp-client.roc @@ -15,7 +15,7 @@ main! : List(OsStr) => Try({}, _) main! = |_args| { stream : Tcp.Stream - stream = Tcp.connect!("127.0.0.1", 8085) ? |err| ConnectFailed(err) + stream = Tcp.connect!("127.0.0.1", 8085, 1_000) ? |err| ConnectFailed(err) verify_stream_methods!(stream)? @@ -27,16 +27,16 @@ main! = |_args| { ## Exercise every read and write operation against the echo test server. verify_stream_methods! : Tcp.Stream => Try({}, _) verify_stream_methods! = |stream| { - stream.write!([1, 2, 3])? - exact_bytes = stream.read_exactly!(3)? + stream.write!([1, 2, 3], 1_000)? + exact_bytes = stream.read_exactly!(3, 1_000)? expect exact_bytes == [1, 2, 3] - stream.write_utf8!("until|")? - until_bytes = stream.read_until!(124)? + stream.write_utf8!("until|", 1_000)? + until_bytes = stream.read_until!(124, 64, 1_000)? expect until_bytes == [117, 110, 116, 105, 108, 124] - stream.write!([42])? - up_to_bytes = stream.read_up_to!(1)? + stream.write!([42], 1_000)? + up_to_bytes = stream.read_up_to!(1, 1_000)? expect up_to_bytes == [42] Ok({}) @@ -51,8 +51,8 @@ run! = |stream| { Err(EndOfFile) => Ok({}) Err(StdinErr(err)) => Err(StdinReadFailed(err)) Ok(out_msg) => { - stream.write_utf8!("${out_msg}\n") ? |err| TcpWriteFailed(err) - in_msg = stream.read_line!() ? |err| TcpReadFailed(err) + stream.write_utf8!("${out_msg}\n", 5_000) ? |err| TcpWriteFailed(err) + in_msg = stream.read_line!(1_048_576, 5_000) ? |err| TcpReadFailed(err) Stdout.line!("< ${in_msg}")? run!(stream) } diff --git a/platform/Host.roc b/platform/Host.roc index 11779003..63933bd3 100644 --- a/platform/Host.roc +++ b/platform/Host.roc @@ -93,11 +93,11 @@ Host :: [].{ stdout_write! : Str => Try({}, [StdoutErr(IOErr)]) stdout_write_bytes! : List(U8) => Try({}, [StdoutErr(IOErr)]) - tcp_connect! : Str, U16 => Try(TcpStream, Str) - tcp_read_up_to! : TcpStream, U64 => Try(List(U8), Str) - tcp_read_exactly! : TcpStream, U64 => Try(List(U8), Str) - tcp_read_until! : TcpStream, U8 => Try(List(U8), Str) - tcp_write! : TcpStream, List(U8) => Try({}, Str) + tcp_connect! : Str, U16, U64 => Try(TcpStream, Str) + tcp_read_up_to! : TcpStream, U64, U64 => Try(List(U8), Str) + tcp_read_exactly! : TcpStream, U64, U64 => Try(List(U8), Str) + tcp_read_until! : TcpStream, U8, U64, U64 => Try(List(U8), Str) + tcp_write! : TcpStream, List(U8), U64 => Try({}, Str) tty_enable_raw_mode! : () => {} tty_disable_raw_mode! : () => {} diff --git a/platform/Tcp.roc b/platform/Tcp.roc index c6c6f884..93b538c9 100644 --- a/platform/Tcp.roc +++ b/platform/Tcp.roc @@ -3,7 +3,12 @@ import Host ## Connect to TCP servers and exchange buffered byte streams. ## ## See the [host runtime behavior](https://github.com/roc-lang/basic-cli#host-runtime-behavior) -## for current timeout and buffering limitations. +## for additional runtime details. +## +## Timeout arguments are in milliseconds and apply to the whole operation, +## including every underlying read/write attempt. A zero timeout fails +## immediately. Read and write timeouts are returned as +## `TcpReadErr(TimedOut)` and `TcpWriteErr(TimedOut)` respectively. Tcp :: [].{ ## Represents a TCP stream. @@ -17,51 +22,57 @@ Tcp :: [].{ to_inspect : Stream -> Str to_inspect = |_| "Tcp.Stream()" - ## Read up to a number of bytes from this TCP stream. - read_up_to! : Stream, U64 => Try(List(U8), _) - read_up_to! = |stream, bytes_to_read| - Host.tcp_read_up_to!(stream.host, bytes_to_read) + ## Read up to a number of bytes, waiting at most `timeout_ms` milliseconds. + read_up_to! : Stream, U64, U64 => Try(List(U8), _) + read_up_to! = |stream, bytes_to_read, timeout_ms| + Host.tcp_read_up_to!(stream.host, bytes_to_read, timeout_ms) .map_err(|err| TcpReadErr(parse_stream_err(err))) - ## Read an exact number of bytes or fail. + ## Read an exact number of bytes, waiting at most `timeout_ms` milliseconds. ## ## `TcpUnexpectedEOF` is returned if the stream ends before the specified ## number of bytes is reached. - read_exactly! : Stream, U64 => Try(List(U8), _) - read_exactly! = |stream, bytes_to_read| - match Host.tcp_read_exactly!(stream.host, bytes_to_read) { + read_exactly! : Stream, U64, U64 => Try(List(U8), _) + read_exactly! = |stream, bytes_to_read, timeout_ms| + match Host.tcp_read_exactly!(stream.host, bytes_to_read, timeout_ms) { Ok(bytes) => Ok(bytes) Err("UnexpectedEof") => Err(TcpUnexpectedEOF) Err(err) => Err(TcpReadErr(parse_stream_err(err))) } - ## Read until a delimiter or EOF is reached. If found, the delimiter is - ## included as the last byte. - read_until! : Stream, U8 => Try(List(U8), _) - read_until! = |stream, byte| - Host.tcp_read_until!(stream.host, byte) - .map_err(|err| TcpReadErr(parse_stream_err(err))) + ## Read until a delimiter or EOF is reached, consuming at most `max_bytes` + ## and waiting at most `timeout_ms` milliseconds. + ## If found, the delimiter is included as the last byte. Returns + ## `TcpReadLimitExceeded(max_bytes)` if the delimiter was not found within + ## the limit. + read_until! : Stream, U8, U64, U64 => Try(List(U8), _) + read_until! = |stream, byte, max_bytes, timeout_ms| + match Host.tcp_read_until!(stream.host, byte, max_bytes, timeout_ms) { + Ok(bytes) => Ok(bytes) + Err("LimitExceeded") => Err(TcpReadLimitExceeded(max_bytes)) + Err(err) => Err(TcpReadErr(parse_stream_err(err))) + } - ## Read until a newline (`\n`, byte 10) or EOF is reached as UTF-8. - ## If found, the newline is included as the last character. - read_line! : Stream => Try(Str, _) - read_line! = |stream| - # NB: use `match` rather than `?` here — `read_until!` yields a - # single-variant error union and `?` currently miscompiles (roc#9826). - match read_until!(stream, 10) { + ## Read at most `max_bytes` through a newline (`\n`, byte 10) or EOF as + ## UTF-8, waiting at most `timeout_ms` milliseconds. If found, the newline + ## is included as the last character. + read_line! : Stream, U64, U64 => Try(Str, _) + read_line! = |stream, max_bytes, timeout_ms| + match read_until!(stream, 10, max_bytes, timeout_ms) { Ok(bytes) => Str.from_utf8(bytes).map_err(|err| TcpReadBadUtf8(err)) Err(err) => Err(err) } - ## Write bytes to this TCP stream. - write! : Stream, List(U8) => Try({}, _) - write! = |stream, bytes| - Host.tcp_write!(stream.host, bytes) + ## Write bytes, waiting at most `timeout_ms` milliseconds. + write! : Stream, List(U8), U64 => Try({}, _) + write! = |stream, bytes, timeout_ms| + Host.tcp_write!(stream.host, bytes, timeout_ms) .map_err(|err| TcpWriteErr(parse_stream_err(err))) - ## Write a string to this TCP stream, encoded as UTF-8. - write_utf8! : Stream, Str => Try({}, _) - write_utf8! = |stream, str| write!(stream, Str.to_utf8(str)) + ## Write a string as UTF-8, waiting at most `timeout_ms` milliseconds. + write_utf8! : Stream, Str, U64 => Try({}, _) + write_utf8! = |stream, str, timeout_ms| + write!(stream, Str.to_utf8(str), timeout_ms) } ## Represents errors that can occur when connecting to a remote host. @@ -83,21 +94,23 @@ Tcp :: [].{ ConnectionRefused, ConnectionReset, Interrupted, + TimedOut, OutOfMemory, BrokenPipe, Unrecognized(Str), ] - ## Opens a TCP connection to a remote host. + ## Opens a TCP connection, waiting at most `timeout_ms` milliseconds across + ## DNS resolution and all resolved addresses. A zero timeout fails immediately. ## ## ```roc ## # Connect to localhost:8080 - ## stream = Tcp.connect!("localhost", 8080)? + ## stream = Tcp.connect!("localhost", 8080, 5_000)? ## ``` ## ## Valid hostnames look like `127.0.0.1`, `::1`, `localhost`, or `roc-lang.org`. - connect! = |host, port| - Host.tcp_connect!(host, port) + connect! = |host, port, timeout_ms| + Host.tcp_connect!(host, port, timeout_ms) .map_ok(|stream| Stream.{ host: stream }) .map_err(parse_connect_err) @@ -122,6 +135,7 @@ Tcp :: [].{ ConnectionRefused => "ConnectionRefused" ConnectionReset => "ConnectionReset" Interrupted => "Interrupted" + TimedOut => "TimedOut" OutOfMemory => "OutOfMemory" BrokenPipe => "BrokenPipe" Unrecognized(message) => "Unrecognized Error: ${message}" @@ -149,7 +163,12 @@ parse_stream_err = |err| "ErrorKind::ConnectionRefused" => ConnectionRefused "ErrorKind::ConnectionReset" => ConnectionReset "ErrorKind::Interrupted" => Interrupted + "ErrorKind::TimedOut" => TimedOut "ErrorKind::OutOfMemory" => OutOfMemory "ErrorKind::BrokenPipe" => BrokenPipe other => Unrecognized(other) } + +expect parse_connect_err("ErrorKind::TimedOut") == TimedOut + +expect parse_stream_err("ErrorKind::TimedOut") == TimedOut diff --git a/scripts/test.py b/scripts/test.py index def91fef..ffcb3ed9 100755 --- a/scripts/test.py +++ b/scripts/test.py @@ -555,6 +555,8 @@ def run_suite( command("cargo", "build", "--locked", "--release", cwd=ROOT / "ci" / "rust_http_server") binaries: dict[str, Path] = {} if "validate" in operations: + print("\n=== RUST TEST ===") + command("cargo", "test", "--locked", "--lib") print("\n=== PLATFORM TEST ===") command("roc", "test", ROOT / "platform" / "main.roc", *roc_extra_args()) for stage in ("fmt", "check", "test"): diff --git a/src/roc_platform_abi.rs b/src/roc_platform_abi.rs index d7043b21..8fd17fe7 100644 --- a/src/roc_platform_abi.rs +++ b/src/roc_platform_abi.rs @@ -118,6 +118,10 @@ pub struct RocErasedCallablePayload { pub on_drop: Option, } +const _: () = assert!(core::mem::size_of::() == 2 * core::mem::size_of::(), "RocErasedCallablePayload size mismatch"); +const _: () = assert!(core::mem::offset_of!(RocErasedCallablePayload, callable_fn_ptr) == 0, "RocErasedCallablePayload.callable_fn_ptr offset mismatch"); +const _: () = assert!(core::mem::offset_of!(RocErasedCallablePayload, on_drop) == core::mem::size_of::(), "RocErasedCallablePayload.on_drop offset mismatch"); + /// Runtime representation of `Box(function)`. pub type RocErasedCallable = *mut u8; @@ -394,6 +398,12 @@ pub struct RocStr { pub length: usize, } +const _: () = assert!(core::mem::size_of::() == 3 * core::mem::size_of::(), "RocStr size mismatch"); +const _: () = assert!(core::mem::align_of::() == core::mem::align_of::(), "RocStr alignment mismatch"); +const _: () = assert!(core::mem::offset_of!(RocStr, bytes) == 0, "RocStr.bytes offset mismatch"); +const _: () = assert!(core::mem::offset_of!(RocStr, capacity_or_alloc_ptr) == core::mem::size_of::(), "RocStr.capacity_or_alloc_ptr offset mismatch"); +const _: () = assert!(core::mem::offset_of!(RocStr, length) == 2 * core::mem::size_of::(), "RocStr.length offset mismatch"); + const ROC_STR_SIZE: usize = core::mem::size_of::(); const ROC_SMALL_STR_MAX_LEN: usize = ROC_STR_SIZE - 1; const ROC_SMALL_STR_BIT: usize = isize::MIN as usize; @@ -595,6 +605,12 @@ pub struct RocListWith { pub capacity_or_alloc_ptr: usize, } +const _: () = assert!(core::mem::size_of::>() == 3 * core::mem::size_of::(), "RocList size mismatch"); +const _: () = assert!(core::mem::align_of::>() == core::mem::align_of::(), "RocList alignment mismatch"); +const _: () = assert!(core::mem::offset_of!(RocList, elements) == 0, "RocList.elements offset mismatch"); +const _: () = assert!(core::mem::offset_of!(RocList, length) == core::mem::size_of::(), "RocList.length offset mismatch"); +const _: () = assert!(core::mem::offset_of!(RocList, capacity_or_alloc_ptr) == 2 * core::mem::size_of::(), "RocList.capacity_or_alloc_ptr offset mismatch"); + impl RocListWith { #[inline] fn header_bytes() -> usize { @@ -4887,53 +4903,59 @@ pub struct HostStdoutWriteBytesArgs { } /// Arguments for Host.tcp_connect! -/// Roc signature: Str, U16 => Try(Host.TcpStream, Str) +/// Roc signature: Str, U16, U64 => Try(Host.TcpStream, Str) /// Refcounted fields are owned by the hosted function. #[repr(C)] #[derive(Clone, Copy)] pub struct HostTcpConnectArgs { pub arg0: RocStr, pub arg1: u16, + pub arg2: u64, } /// Arguments for Host.tcp_read_exactly! -/// Roc signature: Host.TcpStream, U64 => Try(List(U8), Str) +/// Roc signature: Host.TcpStream, U64, U64 => Try(List(U8), Str) /// Refcounted fields are owned by the hosted function. #[repr(C)] #[derive(Clone, Copy)] pub struct HostTcpReadExactlyArgs { pub arg0: *mut u64, pub arg1: u64, + pub arg2: u64, } /// Arguments for Host.tcp_read_until! -/// Roc signature: Host.TcpStream, U8 => Try(List(U8), Str) +/// Roc signature: Host.TcpStream, U8, U64, U64 => Try(List(U8), Str) /// Refcounted fields are owned by the hosted function. #[repr(C)] #[derive(Clone, Copy)] pub struct HostTcpReadUntilArgs { pub arg0: *mut u64, pub arg1: u8, + pub arg2: u64, + pub arg3: u64, } /// Arguments for Host.tcp_read_up_to! -/// Roc signature: Host.TcpStream, U64 => Try(List(U8), Str) +/// Roc signature: Host.TcpStream, U64, U64 => Try(List(U8), Str) /// Refcounted fields are owned by the hosted function. #[repr(C)] #[derive(Clone, Copy)] pub struct HostTcpReadUpToArgs { pub arg0: *mut u64, pub arg1: u64, + pub arg2: u64, } /// Arguments for Host.tcp_write! -/// Roc signature: Host.TcpStream, List(U8) => Try({}, Str) +/// Roc signature: Host.TcpStream, List(U8), U64 => Try({}, Str) /// Refcounted fields are owned by the hosted function. #[repr(C)] #[derive(Clone, Copy)] pub struct HostTcpWriteArgs { pub arg0: *mut u64, pub arg1: RocListWith, + pub arg2: u64, } // Platform Type Aliases @@ -7349,24 +7371,24 @@ unsafe extern "C" { pub fn hosted_stdout_write_bytes(arg0: RocListWith) -> HostStdoutLineResult; /// Hosted symbol for Host.tcp_connect! - /// Roc signature: Str, U16 => Try(Host.TcpStream, Str) - pub fn hosted_tcp_connect(arg0: RocStr, arg1: u16) -> HostTcpConnectResult; + /// Roc signature: Str, U16, U64 => Try(Host.TcpStream, Str) + pub fn hosted_tcp_connect(arg0: RocStr, arg1: u16, arg2: u64) -> HostTcpConnectResult; /// Hosted symbol for Host.tcp_read_exactly! - /// Roc signature: Host.TcpStream, U64 => Try(List(U8), Str) - pub fn hosted_tcp_read_exactly(arg0: *mut u64, arg1: u64) -> HostTcpReadExactlyResult; + /// Roc signature: Host.TcpStream, U64, U64 => Try(List(U8), Str) + pub fn hosted_tcp_read_exactly(arg0: *mut u64, arg1: u64, arg2: u64) -> HostTcpReadExactlyResult; /// Hosted symbol for Host.tcp_read_until! - /// Roc signature: Host.TcpStream, U8 => Try(List(U8), Str) - pub fn hosted_tcp_read_until(arg0: *mut u64, arg1: u8) -> HostTcpReadExactlyResult; + /// Roc signature: Host.TcpStream, U8, U64, U64 => Try(List(U8), Str) + pub fn hosted_tcp_read_until(arg0: *mut u64, arg1: u8, arg2: u64, arg3: u64) -> HostTcpReadExactlyResult; /// Hosted symbol for Host.tcp_read_up_to! - /// Roc signature: Host.TcpStream, U64 => Try(List(U8), Str) - pub fn hosted_tcp_read_up_to(arg0: *mut u64, arg1: u64) -> HostTcpReadExactlyResult; + /// Roc signature: Host.TcpStream, U64, U64 => Try(List(U8), Str) + pub fn hosted_tcp_read_up_to(arg0: *mut u64, arg1: u64, arg2: u64) -> HostTcpReadExactlyResult; /// Hosted symbol for Host.tcp_write! - /// Roc signature: Host.TcpStream, List(U8) => Try({}, Str) - pub fn hosted_tcp_write(arg0: *mut u64, arg1: RocListWith) -> HostTcpWriteResult; + /// Roc signature: Host.TcpStream, List(U8), U64 => Try({}, Str) + pub fn hosted_tcp_write(arg0: *mut u64, arg1: RocListWith, arg2: u64) -> HostTcpWriteResult; /// Hosted symbol for Host.tty_disable_raw_mode! /// Roc signature: {} => {} diff --git a/src/tcp.rs b/src/tcp.rs index 4049b08b..34c71b22 100644 --- a/src/tcp.rs +++ b/src/tcp.rs @@ -1,7 +1,10 @@ use core::mem::ManuallyDrop; use std::ffi::c_void; use std::io::{self, BufRead, BufReader, Read, Write}; -use std::net::TcpStream; +use std::net::{IpAddr, SocketAddr, TcpStream, ToSocketAddrs}; +use std::sync::mpsc; +use std::thread; +use std::time::{Duration, Instant}; use crate::roc_platform_abi::*; use crate::{roc_host, roc_u8_list_from_slice}; @@ -14,8 +17,8 @@ use crate::{roc_host, roc_u8_list_from_slice}; // performs when the stream stays live. // // Errors cross the boundary as a `RocStr` carrying either "ErrorKind::" -// (mapped back to a tag union in Tcp.roc) or "UnexpectedEof"; the Roc side parses -// them into `ConnectErr`/`StreamErr`. +// (mapped back to a tag union in Tcp.roc), "UnexpectedEof", or "LimitExceeded"; +// the Roc side parses them into the public TCP error tags. const TCP_STREAM_BOX_ALIGN: usize = core::mem::align_of::(); @@ -67,7 +70,7 @@ fn to_tcp_connect_err(err: io::Error, roc_host: &RocHost) -> RocStr { io::ErrorKind::AddrNotAvailable => "ErrorKind::AddrNotAvailable".to_string(), io::ErrorKind::ConnectionRefused => "ErrorKind::ConnectionRefused".to_string(), io::ErrorKind::Interrupted => "ErrorKind::Interrupted".to_string(), - io::ErrorKind::TimedOut => "ErrorKind::TimedOut".to_string(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "ErrorKind::TimedOut".to_string(), io::ErrorKind::Unsupported => "ErrorKind::Unsupported".to_string(), other => format!("{:?}", other), }; @@ -80,6 +83,7 @@ fn to_tcp_stream_err(err: io::Error, roc_host: &RocHost) -> RocStr { io::ErrorKind::ConnectionRefused => "ErrorKind::ConnectionRefused".to_string(), io::ErrorKind::ConnectionReset => "ErrorKind::ConnectionReset".to_string(), io::ErrorKind::Interrupted => "ErrorKind::Interrupted".to_string(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "ErrorKind::TimedOut".to_string(), io::ErrorKind::OutOfMemory => "ErrorKind::OutOfMemory".to_string(), io::ErrorKind::BrokenPipe => "ErrorKind::BrokenPipe".to_string(), other => format!("{:?}", other), @@ -87,35 +91,277 @@ fn to_tcp_stream_err(err: io::Error, roc_host: &RocHost) -> RocStr { RocStr::from_str(&message, roc_host) } -// `BufRead::read_until` ported from `roc_file::read_until`, accumulating into a -// plain Vec (the delimiter is included as the last byte when found). -fn tcp_read_until_impl(stream: &mut BufReader, delim: u8) -> io::Result> { +fn timed_out() -> io::Error { + io::Error::new(io::ErrorKind::TimedOut, "TCP operation timed out") +} + +fn deadline_from_timeout(timeout_ms: u64) -> io::Result { + if timeout_ms == 0 { + return Err(timed_out()); + } + + Instant::now() + .checked_add(Duration::from_millis(timeout_ms)) + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "TCP timeout is too large")) +} + +fn remaining_time(deadline: Instant) -> io::Result { + let remaining = deadline + .checked_duration_since(Instant::now()) + .ok_or_else(timed_out)?; + if remaining.is_zero() { + Err(timed_out()) + } else { + Ok(remaining) + } +} + +fn resolve_with_deadline( + host: String, + port: u16, + deadline: Instant, +) -> io::Result> { + if let Ok(ip) = host.parse::() { + return Ok(vec![SocketAddr::new(ip, port)]); + } + + let (sender, receiver) = mpsc::sync_channel(1); + thread::Builder::new() + .name("basic-cli-tcp-resolve".to_string()) + .spawn(move || { + let resolved = (host.as_str(), port) + .to_socket_addrs() + .map(|addresses| addresses.collect()); + let _ = sender.send(resolved); + })?; + + match receiver.recv_timeout(remaining_time(deadline)?) { + Ok(result) => result, + Err(mpsc::RecvTimeoutError::Timeout) => Err(timed_out()), + Err(mpsc::RecvTimeoutError::Disconnected) => Err(io::Error::new( + io::ErrorKind::Other, + "TCP address resolution worker stopped unexpectedly", + )), + } +} + +fn tcp_connect_with_timeout_impl( + host: String, + port: u16, + timeout_ms: u64, +) -> io::Result { + let deadline = deadline_from_timeout(timeout_ms)?; + let addresses = resolve_with_deadline(host, port, deadline)?; + let mut last_error = None; + + for address in addresses { + match TcpStream::connect_timeout(&address, remaining_time(deadline)?) { + Ok(stream) => return Ok(stream), + Err(error) => last_error = Some(error), + } + } + + Err(last_error.unwrap_or_else(|| { + io::Error::new( + io::ErrorKind::AddrNotAvailable, + "hostname resolved to no TCP addresses", + ) + })) +} + +fn set_read_deadline(stream: &TcpStream, deadline: Option) -> io::Result<()> { + if let Some(deadline) = deadline { + stream.set_read_timeout(Some(remaining_time(deadline)?))?; + } + Ok(()) +} + +fn set_write_deadline(stream: &TcpStream, deadline: Option) -> io::Result<()> { + if let Some(deadline) = deadline { + stream.set_write_timeout(Some(remaining_time(deadline)?))?; + } + Ok(()) +} + +fn restore_timeout(result: io::Result, restored: io::Result<()>) -> io::Result { + match result { + Err(error) => Err(error), + Ok(value) => restored.map(|()| value), + } +} + +fn with_read_timeout( + stream: &mut BufReader, + timeout_ms: u64, + operation: impl FnOnce(&mut BufReader, Instant) -> io::Result, +) -> io::Result { + let deadline = deadline_from_timeout(timeout_ms)?; + let previous_timeout = stream.get_ref().read_timeout()?; + let result = operation(stream, deadline); + let restored = stream.get_ref().set_read_timeout(previous_timeout); + restore_timeout(result, restored) +} + +fn tcp_read_up_to_impl( + stream: &mut BufReader, + bytes_to_read: u64, + deadline: Option, +) -> io::Result> { + if bytes_to_read == 0 { + return Ok(Vec::new()); + } + + set_read_deadline(stream.get_ref(), deadline)?; + let available = stream.fill_buf()?; + let limit = usize::try_from(bytes_to_read).unwrap_or(usize::MAX); + let used = available.len().min(limit); + let received = available[..used].to_vec(); + stream.consume(used); + Ok(received) +} + +fn tcp_read_exactly_impl( + stream: &mut BufReader, + bytes_to_read: u64, + deadline: Option, +) -> io::Result> { + let capacity = usize::try_from(bytes_to_read).map_err(|_| { + io::Error::new( + io::ErrorKind::OutOfMemory, + "requested TCP read does not fit in memory", + ) + })?; + let mut buffer = Vec::new(); + buffer.try_reserve_exact(capacity).map_err(|_| { + io::Error::new( + io::ErrorKind::OutOfMemory, + "could not reserve memory for TCP read", + ) + })?; + let mut chunk = [0; 8192]; + + while buffer.len() < capacity { + set_read_deadline(stream.get_ref(), deadline)?; + let remaining = capacity - buffer.len(); + let chunk_len = remaining.min(chunk.len()); + match stream.read(&mut chunk[..chunk_len]) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "TCP stream ended before the requested bytes arrived", + )) + } + Ok(read) => buffer.extend_from_slice(&chunk[..read]), + Err(ref error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + } + + Ok(buffer) +} + +enum ReadUntilError { + Io(io::Error), + LimitExceeded, +} + +impl From for ReadUntilError { + fn from(error: io::Error) -> Self { + Self::Io(error) + } +} + +// `BufRead::read_until` ported from `roc_file::read_until`, with an optional +// caller-selected maximum. The delimiter is included as the last byte when found. +fn tcp_read_until_impl( + stream: &mut BufReader, + delim: u8, + max_bytes: Option, + deadline: Option, +) -> Result, ReadUntilError> { let mut buffer = Vec::new(); loop { - let (done, used) = { + if max_bytes.is_some_and(|limit| buffer.len() as u64 >= limit) { + return Err(ReadUntilError::LimitExceeded); + } + + set_read_deadline(stream.get_ref(), deadline)?; + let (done, used, limit_exceeded) = { let available = match stream.fill_buf() { Ok(n) => n, Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue, - Err(e) => return Err(e), + Err(e) => return Err(ReadUntilError::Io(e)), }; - match available.iter().position(|&b| b == delim) { + let allowed = max_bytes + .map(|limit| usize::try_from(limit - buffer.len() as u64).unwrap_or(usize::MAX)) + .unwrap_or(available.len()) + .min(available.len()); + match available[..allowed].iter().position(|&b| b == delim) { Some(i) => { buffer.extend_from_slice(&available[..=i]); - (true, i + 1) + (true, i + 1, false) } None => { - buffer.extend_from_slice(available); - (false, available.len()) + buffer.extend_from_slice(&available[..allowed]); + (false, allowed, allowed < available.len()) } } }; stream.consume(used); - if done || used == 0 { + if limit_exceeded { + return Err(ReadUntilError::LimitExceeded); + } else if done || used == 0 { return Ok(buffer); } } } +fn tcp_write_impl( + stream: &mut TcpStream, + bytes: &[u8], + deadline: Option, +) -> io::Result<()> { + let mut written = 0; + while written < bytes.len() { + set_write_deadline(stream, deadline)?; + match stream.write(&bytes[written..]) { + Ok(0) => return Err(io::Error::from(io::ErrorKind::WriteZero)), + Ok(count) => written += count, + Err(ref error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) => return Err(error), + } + } + Ok(()) +} + +fn with_write_timeout( + stream: &mut TcpStream, + timeout_ms: u64, + operation: impl FnOnce(&mut TcpStream, Instant) -> io::Result, +) -> io::Result { + let deadline = deadline_from_timeout(timeout_ms)?; + let previous_timeout = stream.write_timeout()?; + let result = operation(stream, deadline); + let restored = stream.set_write_timeout(previous_timeout); + restore_timeout(result, restored) +} + +fn with_read_until_timeout( + stream: &mut BufReader, + delim: u8, + max_bytes: Option, + timeout_ms: u64, +) -> Result, ReadUntilError> { + let deadline = deadline_from_timeout(timeout_ms)?; + let previous_timeout = stream.get_ref().read_timeout()?; + let result = tcp_read_until_impl(stream, delim, max_bytes, Some(deadline)); + let restored = stream.get_ref().set_read_timeout(previous_timeout); + match result { + Err(error) => Err(error), + Ok(value) => restored.map(|()| value).map_err(ReadUntilError::Io), + } +} + fn try_tcp_connect_ok(handle: *mut u64) -> HostTcpConnectResult { HostTcpConnectResult { payload: HostTcpConnectResultPayload { @@ -170,12 +416,16 @@ fn try_tcp_write_err(error: RocStr) -> HostTcpWriteResult { } #[no_mangle] -pub extern "C" fn hosted_tcp_connect(host: RocStr, port: u16) -> HostTcpConnectResult { +pub extern "C" fn hosted_tcp_connect( + host: RocStr, + port: u16, + timeout_ms: u64, +) -> HostTcpConnectResult { let roc_host = roc_host(); let host_string = host.as_str().to_owned(); unsafe { host.decref(roc_host) }; - match TcpStream::connect((host_string.as_str(), port)) { + match tcp_connect_with_timeout_impl(host_string, port, timeout_ms) { Ok(stream) => { let handle = box_tcp_stream(BufReader::new(stream), roc_host); try_tcp_connect_ok(handle) @@ -188,17 +438,15 @@ pub extern "C" fn hosted_tcp_connect(host: RocStr, port: u16) -> HostTcpConnectR pub extern "C" fn hosted_tcp_read_up_to( handle: *mut u64, bytes_to_read: u64, + timeout_ms: u64, ) -> HostTcpReadUpToResult { let roc_host = roc_host(); let result = { let stream = unsafe { tcp_stream_ref(handle) }; - let mut chunk = stream.take(bytes_to_read); - match chunk.fill_buf() { - Ok(received) => { - let received = received.to_vec(); - stream.consume(received.len()); - try_tcp_read_ok(roc_u8_list_from_slice(&received, roc_host)) - } + match with_read_timeout(stream, timeout_ms, |stream, deadline| { + tcp_read_up_to_impl(stream, bytes_to_read, Some(deadline)) + }) { + Ok(received) => try_tcp_read_ok(roc_u8_list_from_slice(&received, roc_host)), Err(err) => try_tcp_read_err(to_tcp_stream_err(err, roc_host)), } }; @@ -210,19 +458,17 @@ pub extern "C" fn hosted_tcp_read_up_to( pub extern "C" fn hosted_tcp_read_exactly( handle: *mut u64, bytes_to_read: u64, + timeout_ms: u64, ) -> HostTcpReadExactlyResult { let roc_host = roc_host(); let result = { let stream = unsafe { tcp_stream_ref(handle) }; - let mut buffer = Vec::with_capacity(bytes_to_read as usize); - let mut chunk = stream.take(bytes_to_read); - match chunk.read_to_end(&mut buffer) { - Ok(read) => { - if (read as u64) < bytes_to_read { - try_tcp_read_err(RocStr::from_str("UnexpectedEof", roc_host)) - } else { - try_tcp_read_ok(roc_u8_list_from_slice(&buffer, roc_host)) - } + match with_read_timeout(stream, timeout_ms, |stream, deadline| { + tcp_read_exactly_impl(stream, bytes_to_read, Some(deadline)) + }) { + Ok(buffer) => try_tcp_read_ok(roc_u8_list_from_slice(&buffer, roc_host)), + Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => { + try_tcp_read_err(RocStr::from_str("UnexpectedEof", roc_host)) } Err(err) => try_tcp_read_err(to_tcp_stream_err(err, roc_host)), } @@ -232,13 +478,21 @@ pub extern "C" fn hosted_tcp_read_exactly( } #[no_mangle] -pub extern "C" fn hosted_tcp_read_until(handle: *mut u64, byte: u8) -> HostTcpReadUntilResult { +pub extern "C" fn hosted_tcp_read_until( + handle: *mut u64, + byte: u8, + max_bytes: u64, + timeout_ms: u64, +) -> HostTcpReadUntilResult { let roc_host = roc_host(); let result = { let stream = unsafe { tcp_stream_ref(handle) }; - match tcp_read_until_impl(stream, byte) { + match with_read_until_timeout(stream, byte, Some(max_bytes), timeout_ms) { Ok(buffer) => try_tcp_read_ok(roc_u8_list_from_slice(&buffer, roc_host)), - Err(err) => try_tcp_read_err(to_tcp_stream_err(err, roc_host)), + Err(ReadUntilError::Io(err)) => try_tcp_read_err(to_tcp_stream_err(err, roc_host)), + Err(ReadUntilError::LimitExceeded) => { + try_tcp_read_err(RocStr::from_str("LimitExceeded", roc_host)) + } } }; release_tcp_stream(handle, roc_host); @@ -249,11 +503,14 @@ pub extern "C" fn hosted_tcp_read_until(handle: *mut u64, byte: u8) -> HostTcpRe pub extern "C" fn hosted_tcp_write( handle: *mut u64, msg: RocListWith, + timeout_ms: u64, ) -> HostTcpWriteResult { let roc_host = roc_host(); let result = { let stream = unsafe { tcp_stream_ref(handle) }; - match stream.get_mut().write_all(msg.as_slice()) { + match with_write_timeout(stream.get_mut(), timeout_ms, |stream, deadline| { + tcp_write_impl(stream, msg.as_slice(), Some(deadline)) + }) { Ok(()) => try_tcp_write_ok(), Err(err) => try_tcp_write_err(to_tcp_stream_err(err, roc_host)), } @@ -262,3 +519,135 @@ pub extern "C" fn hosted_tcp_write( release_tcp_stream(handle, roc_host); result } + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{Shutdown, TcpListener}; + + fn local_server( + handler: impl FnOnce(TcpStream) + Send + 'static, + ) -> (u16, thread::JoinHandle<()>) { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + handler(stream); + }); + (port, server) + } + + #[test] + fn bounded_delimiter_read_stops_at_the_limit() { + let (port, server) = local_server(|mut stream| { + stream.write_all(b"delimiter-free").unwrap(); + stream.shutdown(Shutdown::Write).unwrap(); + }); + let stream = TcpStream::connect(("127.0.0.1", port)).unwrap(); + let mut reader = BufReader::new(stream); + + assert!(matches!( + tcp_read_until_impl(&mut reader, b'|', Some(4), None), + Err(ReadUntilError::LimitExceeded) + )); + server.join().unwrap(); + } + + #[test] + fn stalled_read_respects_the_deadline() { + let (port, server) = local_server(|_stream| { + thread::sleep(Duration::from_millis(150)); + }); + let stream = TcpStream::connect(("127.0.0.1", port)).unwrap(); + let mut reader = BufReader::new(stream); + + let error = with_read_timeout(&mut reader, 20, |stream, deadline| { + tcp_read_up_to_impl(stream, 1, Some(deadline)) + }) + .unwrap_err(); + assert!(matches!( + error.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + )); + server.join().unwrap(); + } + + #[test] + fn exact_read_reports_unexpected_eof() { + let (port, server) = local_server(|mut stream| { + stream.write_all(b"hi").unwrap(); + stream.shutdown(Shutdown::Write).unwrap(); + }); + let stream = TcpStream::connect(("127.0.0.1", port)).unwrap(); + let mut reader = BufReader::new(stream); + + let error = tcp_read_exactly_impl(&mut reader, 3, None).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof); + server.join().unwrap(); + } + + #[test] + fn connect_timeout_includes_localhost_resolution() { + // Bind through the same hostname the client resolves so both sides use + // the platform's preferred address family (IPv6 on Windows, IPv4 on + // some Unix hosts). Otherwise an unavailable first address can consume + // a deliberately short deadline before the matching address is tried. + let listener = TcpListener::bind(("localhost", 0)).unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + listener.accept().unwrap(); + }); + + tcp_connect_with_timeout_impl("localhost".to_string(), port, 10_000).unwrap(); + server.join().unwrap(); + + let error = tcp_connect_with_timeout_impl("localhost".to_string(), port, 0).unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + } + + #[cfg(target_os = "linux")] + #[test] + fn stalled_connect_respects_the_deadline() { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let address = listener.local_addr().unwrap(); + let mut queued_connections = Vec::new(); + + loop { + match TcpStream::connect_timeout(&address, Duration::from_millis(10)) { + Ok(stream) => queued_connections.push(stream), + Err(error) + if matches!( + error.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + ) => + { + break; + } + Err(error) => panic!("could not fill the local TCP accept queue: {error}"), + } + assert!(queued_connections.len() < 4_096); + } + + let started = Instant::now(); + let error = tcp_connect_with_timeout_impl(address.ip().to_string(), address.port(), 20) + .unwrap_err(); + assert!(matches!( + error.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + )); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[test] + fn zero_write_timeout_fails_before_writing() { + let (port, server) = local_server(|_stream| {}); + let mut stream = TcpStream::connect(("127.0.0.1", port)).unwrap(); + + let error = with_write_timeout(&mut stream, 0, |stream, deadline| { + tcp_write_impl(stream, b"hello", Some(deadline)) + }) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + server.join().unwrap(); + } +}