Add TCP timeouts and bounded delimiter reads

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

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)
 		}

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! : () => {}

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(<opaque>)"
 
-		## 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

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"):

src/roc_platform_abi.rs

@@ -118,6 +118,10 @@ pub struct RocErasedCallablePayload {
     pub on_drop: Option<RocErasedCallableOnDrop>,
 }
 
+const _: () = assert!(core::mem::size_of::<RocErasedCallablePayload>() == 2 * core::mem::size_of::<usize>(), "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::<usize>(), "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::<RocStr>() == 3 * core::mem::size_of::<usize>(), "RocStr size mismatch");
+const _: () = assert!(core::mem::align_of::<RocStr>() == core::mem::align_of::<usize>(), "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::<usize>(), "RocStr.capacity_or_alloc_ptr offset mismatch");
+const _: () = assert!(core::mem::offset_of!(RocStr, length) == 2 * core::mem::size_of::<usize>(), "RocStr.length offset mismatch");
+
 const ROC_STR_SIZE: usize = core::mem::size_of::<RocStr>();
 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<T, const ELEMENTS_REFCOUNTED: bool> {
     pub capacity_or_alloc_ptr: usize,
 }
 
+const _: () = assert!(core::mem::size_of::<RocList<u8>>() == 3 * core::mem::size_of::<usize>(), "RocList size mismatch");
+const _: () = assert!(core::mem::align_of::<RocList<u8>>() == core::mem::align_of::<usize>(), "RocList alignment mismatch");
+const _: () = assert!(core::mem::offset_of!(RocList<u8>, elements) == 0, "RocList.elements offset mismatch");
+const _: () = assert!(core::mem::offset_of!(RocList<u8>, length) == core::mem::size_of::<usize>(), "RocList.length offset mismatch");
+const _: () = assert!(core::mem::offset_of!(RocList<u8>, capacity_or_alloc_ptr) == 2 * core::mem::size_of::<usize>(), "RocList.capacity_or_alloc_ptr offset mismatch");
+
 impl<T, const ELEMENTS_REFCOUNTED: bool> RocListWith<T, ELEMENTS_REFCOUNTED> {
     #[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<u8, false>,
+    pub arg2: u64,
 }
 
 // Platform Type Aliases
@@ -7349,24 +7371,24 @@ unsafe extern "C" {
     pub fn hosted_stdout_write_bytes(arg0: RocListWith<u8, false>) -> 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<u8, false>) -> HostTcpWriteResult;
+    /// Roc signature: Host.TcpStream, List(U8), U64 => Try({}, Str)
+    pub fn hosted_tcp_write(arg0: *mut u64, arg1: RocListWith<u8, false>, arg2: u64) -> HostTcpWriteResult;
 
     /// Hosted symbol for Host.tty_disable_raw_mode!
     /// Roc signature: {} => {}