v0.2.6: preserve buffered lines and add request read timeouts

8 files changed, 203 insertions, 23 deletions

diff --git a/Cargo.toml b/Cargo.toml
index 8228349..359d16f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "zagan-bet"
-version = "0.2.5"
+version = "0.2.6"
 edition = "2024"
 
 [dependencies]
diff --git a/src/bin/client/connection.rs b/src/bin/client/connection.rs
index 93a7829..49695dd 100644
--- a/src/bin/client/connection.rs
+++ b/src/bin/client/connection.rs
@@ -66,6 +66,7 @@ pub fn connect(
     }
 
     let mut tls = tls;
+    tls.set_read_timeout(Some(Duration::from_secs(10)))?;
 
     let start = Instant::now();
     tls.write_line("PING")?;
@@ -124,6 +125,13 @@ impl TlsStream {
         self.inner.write_line(msg)
     }
 
+    pub fn set_read_timeout(
+        &self,
+        duration: Option<std::time::Duration>,
+    ) -> std::io::Result<()> {
+        self.inner.set_read_timeout(duration)
+    }
+
     pub fn try_clone(&self) -> std::io::Result<TlsStream> {
         Ok(TlsStream {
             inner: self.inner.try_clone()?,
diff --git a/src/bin/client/repl.rs b/src/bin/client/repl.rs
index 76b149b..033fb1a 100644
--- a/src/bin/client/repl.rs
+++ b/src/bin/client/repl.rs
@@ -18,6 +18,7 @@ use zagan_bet::client::database;
 use zagan_bet::client::ui::{
     display_width, format_pong, host_lines, input_cursor_column, lines_to_move_up,
 };
+use zagan_bet::protocol::io_failure_message;
 
 use crate::connection::TlsStream;
 
@@ -299,9 +300,9 @@ impl ClientState {
                 self.stream = Some(stream);
                 response
             }
-            Err(_) => {
+            Err(error) => {
                 self.stream = None;
-                "ERROR Connection lost".to_string()
+                io_failure_message(&error).to_string()
             }
         }
     }
diff --git a/src/bin/client/tls.rs b/src/bin/client/tls.rs
index ca99838..b118f63 100644
--- a/src/bin/client/tls.rs
+++ b/src/bin/client/tls.rs
@@ -19,6 +19,7 @@ use std::sync::Arc;
 use parking_lot::Mutex;
 use rustls::client::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
 use rustls::DigitallySignedStruct;
+use zagan_bet::protocol::LineBuffer;
 use zagan_bet::tofu::{decide, spki_fingerprint, TrustDecision};
 
 pub struct TlsConnector {
@@ -59,7 +60,11 @@ impl TlsConnector {
         conn.complete_io(&mut sock)?;
 
         Ok(TlsStream {
-            session: Arc::new(Mutex::new(Session { sock, tls: conn })),
+            session: Arc::new(Mutex::new(Session {
+                sock,
+                tls: conn,
+                lines: LineBuffer::new(),
+            })),
         })
     }
 
@@ -144,6 +149,7 @@ impl ServerCertVerifier for TofuVerifier {
 struct Session {
     sock: TcpStream,
     tls: rustls::ClientConnection,
+    lines: LineBuffer,
 }
 
 impl Session {
@@ -159,17 +165,25 @@ pub struct TlsStream {
 
 impl TlsStream {
     pub fn read_line(&mut self, buf: &mut String) -> std::io::Result<usize> {
-        let mut byte_buf = [0u8; 1024];
         let mut s = self.session.lock();
         loop {
+            if let Some(line) = s.lines.next_line() {
+                buf.push_str(&line);
+                buf.push('\n');
+                return Ok(line.len() + 1);
+            }
+
+            let mut byte_buf = [0u8; 4096];
             match s.tls.reader().read(&mut byte_buf) {
-                Ok(0) => return Ok(0),
-                Ok(n) => {
-                    buf.push_str(std::str::from_utf8(&byte_buf[..n]).unwrap_or(""));
-                    if buf.contains('\n') {
-                        return Ok(n);
+                Ok(0) => {
+                    if let Some(rest) = s.lines.take_rest() {
+                        let read = rest.len();
+                        buf.push_str(&rest);
+                        return Ok(read);
                     }
+                    return Ok(0);
                 }
+                Ok(n) => s.lines.push(&byte_buf[..n]),
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => s.drive()?,
                 Err(e) => return Err(e),
             }
@@ -183,6 +197,14 @@ impl TlsStream {
         s.drive()
     }
 
+    /// Apply a read timeout to the underlying TCP socket.
+    pub fn set_read_timeout(
+        &self,
+        duration: Option<std::time::Duration>,
+    ) -> std::io::Result<()> {
+        self.session.lock().sock.set_read_timeout(duration)
+    }
+
     /// Clones share the same underlying TLS session.
     pub fn try_clone(&self) -> std::io::Result<TlsStream> {
         Ok(self.clone())
diff --git a/src/bin/server/tls.rs b/src/bin/server/tls.rs
index acf3315..62a25db 100644
--- a/src/bin/server/tls.rs
+++ b/src/bin/server/tls.rs
@@ -1,6 +1,8 @@
 use std::io::{Read, Write};
 use std::net::TcpStream;
 
+use zagan_bet::protocol::LineBuffer;
+
 #[derive(Clone)]
 pub struct TlsServer {
     acceptor: native_tls::TlsAcceptor,
@@ -42,27 +44,25 @@ impl TlsServer {
 
 pub struct TlsStream {
     stream: native_tls::TlsStream<TcpStream>,
+    lines: LineBuffer,
 }
 
 impl TlsStream {
     pub fn new(stream: native_tls::TlsStream<TcpStream>) -> Self {
-        Self { stream }
+        Self {
+            stream,
+            lines: LineBuffer::new(),
+        }
     }
 
     pub fn read_line(&mut self, s: &mut String) -> std::io::Result<usize> {
-        let mut buf = [0u8; 1024];
-        loop {
-            match self.stream.read(&mut buf) {
-                Ok(0) => return Ok(0),
-                Ok(n) => {
-                    let chunk = std::str::from_utf8(&buf[..n]).unwrap_or("");
-                    s.push_str(chunk);
-                    if s.contains('\n') {
-                        return Ok(n);
-                    }
-                }
-                Err(e) => return Err(e),
+        match self.lines.read_line(&mut self.stream)? {
+            Some(line) => {
+                s.push_str(&line);
+                s.push('\n');
+                Ok(line.len() + 1)
             }
+            None => Ok(0),
         }
     }
 
diff --git a/src/lib.rs b/src/lib.rs
index a4fb801..e700173 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -6,5 +6,6 @@
 //! `tests/` without a network or a terminal.
 
 pub mod client;
+pub mod protocol;
 pub mod server;
 pub mod tofu;
diff --git a/src/protocol.rs b/src/protocol.rs
new file mode 100644
index 0000000..11f54a1
--- /dev/null
+++ b/src/protocol.rs
@@ -0,0 +1,70 @@
+//! Newline-delimited protocol helpers shared by the client and server.
+//!
+//! TLS reads can return more than one line at a time. Keeping the leftover
+//! bytes is essential: dropping them silently loses the next message.
+
+use std::io::{self, Read};
+
+#[derive(Debug, Default)]
+pub struct LineBuffer {
+    buffer: Vec<u8>,
+}
+
+impl LineBuffer {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn push(&mut self, bytes: &[u8]) {
+        self.buffer.extend_from_slice(bytes);
+    }
+
+    /// Return the next complete line, including a line that ended at EOF.
+    pub fn next_line(&mut self) -> Option<String> {
+        let end = self.buffer.iter().position(|byte| *byte == b'\n')?;
+        let mut line: Vec<u8> = self.buffer.drain(..=end).collect();
+        line.pop(); // '\n'
+        if line.last() == Some(&b'\r') {
+            line.pop();
+        }
+        Some(String::from_utf8_lossy(&line).into_owned())
+    }
+
+    /// Return the unterminated remainder at EOF, if any.
+    pub fn take_rest(&mut self) -> Option<String> {
+        if self.buffer.is_empty() {
+            return None;
+        }
+        let mut rest = std::mem::take(&mut self.buffer);
+        if rest.last() == Some(&b'\r') {
+            rest.pop();
+        }
+        Some(String::from_utf8_lossy(&rest).into_owned())
+    }
+
+    /// Read one line from a blocking reader, preserving any extra lines for
+    /// the next call. Returns `None` at EOF with nothing buffered.
+    pub fn read_line<R: Read>(&mut self, reader: &mut R) -> io::Result<Option<String>> {
+        loop {
+            if let Some(line) = self.next_line() {
+                return Ok(Some(line));
+            }
+
+            let mut chunk = [0u8; 4096];
+            match reader.read(&mut chunk) {
+                Ok(0) => return Ok(self.take_rest()),
+                Ok(n) => self.push(&chunk[..n]),
+                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
+                Err(error) => return Err(error),
+            }
+        }
+    }
+}
+
+/// A stable, user-facing message for a failed request.
+pub fn io_failure_message(error: &io::Error) -> &'static str {
+    match error.kind() {
+        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "ERROR Request timed out",
+        _ => "ERROR Connection lost",
+    }
+}
diff --git a/tests/protocol.rs b/tests/protocol.rs
new file mode 100644
index 0000000..2962574
--- /dev/null
+++ b/tests/protocol.rs
@@ -0,0 +1,78 @@
+use std::io::{Cursor, Error, ErrorKind};
+
+use zagan_bet::protocol::{io_failure_message, LineBuffer};
+
+#[test]
+fn keeps_data_after_the_first_newline() {
+    let mut buffer = LineBuffer::new();
+    buffer.push(b"PONG\nGAME_RESULT session 60 10 990 OSS two\n");
+
+    assert_eq!(buffer.next_line().as_deref(), Some("PONG"));
+    assert_eq!(
+        buffer.next_line().as_deref(),
+        Some("GAME_RESULT session 60 10 990 OSS two")
+    );
+    assert_eq!(buffer.next_line(), None);
+}
+
+#[test]
+fn handles_crlf_and_empty_lines() {
+    let mut buffer = LineBuffer::new();
+    buffer.push(b"PONG\r\n\r\n");
+
+    assert_eq!(buffer.next_line().as_deref(), Some("PONG"));
+    assert_eq!(buffer.next_line().as_deref(), Some(""));
+    assert_eq!(buffer.next_line(), None);
+}
+
+#[test]
+fn waits_for_a_split_line() {
+    let mut buffer = LineBuffer::new();
+    buffer.push(b"PO");
+    assert_eq!(buffer.next_line(), None);
+
+    buffer.push(b"NG\n");
+    assert_eq!(buffer.next_line().as_deref(), Some("PONG"));
+}
+
+#[test]
+fn reads_multiple_lines_from_a_reader() {
+    let mut buffer = LineBuffer::new();
+    let mut reader = Cursor::new(b"one\ntwo\n".to_vec());
+
+    assert_eq!(buffer.read_line(&mut reader).unwrap().as_deref(), Some("one"));
+    assert_eq!(buffer.read_line(&mut reader).unwrap().as_deref(), Some("two"));
+    assert_eq!(buffer.read_line(&mut reader).unwrap(), None);
+}
+
+#[test]
+fn returns_a_partial_line_at_eof() {
+    let mut buffer = LineBuffer::new();
+    let mut reader = Cursor::new(b"partial".to_vec());
+
+    assert_eq!(
+        buffer.read_line(&mut reader).unwrap().as_deref(),
+        Some("partial")
+    );
+    assert_eq!(buffer.read_line(&mut reader).unwrap(), None);
+}
+
+#[test]
+fn timeout_errors_get_a_timeout_message() {
+    assert_eq!(
+        io_failure_message(&Error::from(ErrorKind::TimedOut)),
+        "ERROR Request timed out"
+    );
+    assert_eq!(
+        io_failure_message(&Error::from(ErrorKind::WouldBlock)),
+        "ERROR Request timed out"
+    );
+    assert_eq!(
+        io_failure_message(&Error::from(ErrorKind::UnexpectedEof)),
+        "ERROR Connection lost"
+    );
+    assert_eq!(
+        io_failure_message(&Error::from(ErrorKind::ConnectionReset)),
+        "ERROR Connection lost"
+    );
+}