1 files changed, 109 insertions, 62 deletions

diff --git a/src/client/transport.rs b/src/client/transport.rs
index 97c9227..60e7ab6 100644
--- a/src/client/transport.rs
+++ b/src/client/transport.rs
@@ -7,20 +7,27 @@
 use std::fmt;
 use std::io::{Read, Write};
 use std::net::TcpStream;
-use std::sync::mpsc;
 use std::sync::Arc;
+use std::sync::mpsc;
 use std::thread;
 use std::time::{Duration, Instant};
 
 use parking_lot::Mutex;
-use rustls::client::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
-use rustls::DigitallySignedStruct;
 use rusqlite::Connection;
+use rustls::DigitallySignedStruct;
+use rustls::client::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
 
 use crate::client::database;
-use crate::protocol::LineBuffer;
-use crate::tofu::{decide, spki_fingerprint, TrustDecision};
-use crate::version::{parse_hello, HelloReply, VERSION};
+use crate::protocol::{
+    FrameBuffer, Packet, parse_packet, request_line, write_frame as write_protocol_frame,
+};
+use crate::tofu::{TrustDecision, decide, spki_fingerprint};
+use crate::version::{HelloReply, VERSION, parse_hello};
+
+/// Reserved request ids for the transport handshake. Application requests
+/// start at 1, so the two ranges cannot collide on one connection.
+const HELLO_REQUEST_ID: u64 = u64::MAX;
+const AUTH_REQUEST_ID: u64 = u64::MAX - 1;
 
 /// How long to wait for TCP + TLS before giving up.
 const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
@@ -43,6 +50,8 @@ pub struct Connected {
     pub server_version: String,
     pub latest: String,
     pub url: String,
+    /// Server Unix time reported by `HELLO`, milliseconds.
+    pub server_time: u64,
     /// Round trip of the HELLO exchange.
     pub elapsed: Duration,
 }
@@ -51,10 +60,7 @@ pub struct Connected {
 pub enum TransportError {
     Timeout,
     TrustDeclined,
-    KeyChanged {
-        expected: String,
-        observed: String,
-    },
+    KeyChanged { expected: String, observed: String },
     VersionMismatch(String),
     Closed,
     Unexpected(String),
@@ -121,7 +127,7 @@ pub fn connect(
                 return Err(TransportError::KeyChanged {
                     expected,
                     observed: actual,
-                })
+                });
             }
         }
     }
@@ -131,12 +137,7 @@ pub fn connect(
         .map_err(io_failure)?;
 
     let start = Instant::now();
-    tls.write_line(&format!("HELLO {VERSION}"))
-        .map_err(io_failure)?;
-    let mut response = String::new();
-    if tls.read_line(&mut response).map_err(io_failure)? == 0 {
-        return Err(TransportError::Closed);
-    }
+    let response = exchange(&mut tls, HELLO_REQUEST_ID, &format!("HELLO {VERSION}"))?;
     let elapsed = start.elapsed();
 
     let response = response.trim();
@@ -151,10 +152,11 @@ pub fn connect(
     let mut user = None;
     let mut auth_error = None;
     if let Some((username, password)) = database::get_credentials(conn, &observed) {
-        tls.write_line(&format!("AUTH {username} {password}"))
-            .map_err(io_failure)?;
-        let mut auth = String::new();
-        tls.read_line(&mut auth).map_err(io_failure)?;
+        let auth = exchange(
+            &mut tls,
+            AUTH_REQUEST_ID,
+            &format!("AUTH {username} {password}"),
+        )?;
         if auth.trim().starts_with("OK") {
             user = Some(username);
         } else {
@@ -169,19 +171,31 @@ pub fn connect(
         server_version: hello.version,
         latest: hello.latest,
         url: hello.url,
+        server_time: hello.time,
         elapsed,
     })
 }
 
+/// Send one request and wait for the reply with the same id, ignoring events
+/// that arrive in between. Used only for the handshake.
+fn exchange(stream: &mut TlsStream, id: u64, payload: &str) -> Result<String, TransportError> {
+    stream
+        .write_request(id, None, payload)
+        .map_err(io_failure)?;
+    loop {
+        match stream.read_packet().map_err(io_failure)? {
+            None => return Err(TransportError::Closed),
+            Some(Packet::Reply { id: reply_id, body }) if reply_id == id => return Ok(body),
+            Some(Packet::Reply { .. }) | Some(Packet::Event { .. }) => continue,
+        }
+    }
+}
+
 /// TCP + TLS on a thread so a blocked handshake can time out.
 fn tls_connect(addr: &str) -> Result<(TlsStream, Option<String>), TransportError> {
     let (sender, receiver) = mpsc::channel();
     let addr = addr.to_string();
-    let server_name = addr
-        .split(':')
-        .next()
-        .unwrap_or("127.0.0.1")
-        .to_string();
+    let server_name = addr.split(':').next().unwrap_or("127.0.0.1").to_string();
 
     thread::spawn(move || {
         let outcome = (|| -> Result<(TlsStream, Option<String>), TransportError> {
@@ -241,7 +255,7 @@ impl TlsConnector {
             session: Arc::new(Mutex::new(Session {
                 sock,
                 tls: connection,
-                lines: LineBuffer::new(),
+                frames: FrameBuffer::new(),
                 read_timeout: Some(READ_TIMEOUT),
             })),
         })
@@ -268,8 +282,9 @@ impl ServerCertVerifier for RecordingVerifier {
         _ocsp_response: &[u8],
         _now: std::time::SystemTime,
     ) -> Result<ServerCertVerified, rustls::Error> {
-        let fingerprint = spki_fingerprint(&end_entity.0)
-            .map_err(|error| rustls::Error::General(format!("certificate parse failed: {error}")))?;
+        let fingerprint = spki_fingerprint(&end_entity.0).map_err(|error| {
+            rustls::Error::General(format!("certificate parse failed: {error}"))
+        })?;
         *self.observed.lock() = Some(fingerprint);
         Ok(ServerCertVerified::assertion())
     }
@@ -313,7 +328,7 @@ impl ServerCertVerifier for RecordingVerifier {
 struct Session {
     sock: TcpStream,
     tls: rustls::ClientConnection,
-    lines: LineBuffer,
+    frames: FrameBuffer,
     read_timeout: Option<Duration>,
 }
 
@@ -329,37 +344,58 @@ pub struct TlsStream {
 }
 
 impl TlsStream {
-    pub fn read_line(&mut self, buf: &mut String) -> std::io::Result<usize> {
+    /// Write one enveloped request. Application code allocates `id`.
+    pub fn write_request(
+        &mut self,
+        id: u64,
+        idem: Option<&str>,
+        payload: &str,
+    ) -> std::io::Result<()> {
+        self.write_frame(&request_line(id, idem, payload))
+    }
+
+    /// Write one raw frame payload.
+    pub fn write_frame(&mut self, payload: &str) -> std::io::Result<()> {
+        let mut s = self.session.lock();
+        write_protocol_frame(&mut s.tls.writer(), payload)?;
+        s.drive()
+    }
+
+    /// Block until one whole frame arrives.
+    pub fn read_frame(&mut self) -> std::io::Result<Option<String>> {
         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);
+            match s.frames.next_frame() {
+                Some(Ok(frame)) => return Ok(Some(frame)),
+                Some(Err(error)) => {
+                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error));
+                }
+                None => {}
             }
 
             let mut chunk = [0u8; 4096];
             match s.tls.reader().read(&mut chunk) {
-                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(&chunk[..n]),
+                Ok(0) => return Ok(None),
+                Ok(n) => s.frames.push(&chunk[..n]),
                 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => s.drive()?,
                 Err(error) => return Err(error),
             }
         }
     }
 
-    pub fn write_line(&mut self, message: &str) -> std::io::Result<()> {
-        let mut s = self.session.lock();
-        s.tls.writer().write_all(message.as_bytes())?;
-        s.tls.writer().write_all(b"\n")?;
-        s.drive()
+    /// Block until one `RES` or `EVT` arrives.
+    pub fn read_packet(&mut self) -> std::io::Result<Option<Packet>> {
+        loop {
+            let Some(frame) = self.read_frame()? else {
+                return Ok(None);
+            };
+            match parse_packet(&frame) {
+                Ok(packet) => return Ok(Some(packet)),
+                Err(error) => {
+                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error));
+                }
+            }
+        }
     }
 
     pub fn set_read_timeout(&self, duration: Option<Duration>) -> std::io::Result<()> {
@@ -368,24 +404,36 @@ impl TlsStream {
         s.sock.set_read_timeout(duration)
     }
 
-    /// Return one complete line if the socket already has one, without
-    /// blocking. Used to pick up server pushes between requests.
-    pub fn read_available_line(&mut self) -> std::io::Result<Option<String>> {
+    /// Return one packet if the socket already has one, without blocking.
+    /// Used to pick up server pushes between requests.
+    pub fn read_available_packet(&mut self) -> std::io::Result<Option<Packet>> {
         let mut s = self.session.lock();
-        if let Some(line) = s.lines.next_line() {
-            return Ok(Some(line));
-        }
 
         // `set_read_timeout(Some(0))` is rejected by std, so poll with the
         // shortest timeout the socket accepts instead of a nonblocking mode.
         s.sock
             .set_read_timeout(Some(std::time::Duration::from_millis(1)))?;
-        let drained = (|| -> std::io::Result<()> {
+        let drained = (|| -> std::io::Result<Option<Packet>> {
             loop {
+                match s.frames.next_frame() {
+                    Some(Ok(frame)) => {
+                        return match parse_packet(&frame) {
+                            Ok(packet) => Ok(Some(packet)),
+                            Err(error) => {
+                                Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error))
+                            }
+                        };
+                    }
+                    Some(Err(error)) => {
+                        return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, error));
+                    }
+                    None => {}
+                }
+
                 let mut buf = [0u8; 4096];
                 match s.tls.reader().read(&mut buf) {
-                    Ok(0) => return Ok(()),
-                    Ok(n) => s.lines.push(&buf[..n]),
+                    Ok(0) => return Ok(None),
+                    Ok(n) => s.frames.push(&buf[..n]),
                     // The reader only drains plaintext rustls has already
                     // decrypted; `drive` moves bytes off the socket.
                     Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
@@ -397,7 +445,7 @@ impl TlsStream {
                                     std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                                 ) =>
                             {
-                                return Ok(());
+                                return Ok(None);
                             }
                             Err(error) => return Err(error),
                         }
@@ -408,15 +456,14 @@ impl TlsStream {
                             std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                         ) =>
                     {
-                        return Ok(());
+                        return Ok(None);
                     }
                     Err(error) => return Err(error),
                 }
             }
         })();
         s.sock.set_read_timeout(s.read_timeout)?;
-        drained?;
-        Ok(s.lines.next_line())
+        drained
     }
 }