Implement PTY-native session terminal control

crates/lazaret-client/src/lib.rs

@@ -1,8 +1,10 @@
 use std::fmt;
 use std::fs::File;
 use std::io::{self, Write};
+use std::net::Shutdown;
 use std::path::PathBuf;
-use std::sync::Mutex;
+use std::sync::{Mutex, mpsc};
+use std::thread::JoinHandle;
 use std::time::Duration;
 
 use ed25519_dalek::SigningKey;
@@ -10,8 +12,8 @@ use lazaret_control_client::{ControlClient, ControlClientError, PendingClientEnr
 use lazaret_core::identity::{ClientId, SessionAnchorId, SessionId};
 use lazaret_core::records::{
     ClientEnrollmentResponse, ControlRequest, ControlResponse, DeveloperDecision,
-    DeveloperDecisionKind, InteractiveSessionBatch, InteractiveSessionEvent,
-    InteractiveSessionRequest, NonInteractiveSessionMessage, NonInteractiveSessionRequest,
+    DeveloperDecisionKind, InteractiveSessionBatchV2, InteractiveSessionEventV2,
+    InteractiveSessionRequestV2, NonInteractiveSessionMessage, NonInteractiveSessionRequest,
     ProtocolPublicationIdentity, ProtocolVersion, SessionBinding, SessionExitOutcome,
     SessionRegistrationRequest, TlsSubjectId,
 };
@@ -414,9 +416,9 @@ pub fn execute_interactive_over_control_face_as<E, H: HostTerminal>(
     let cli_id = identity.cli_id;
     let session_id = identity.session_id;
     let pending = PendingClientEnrollment::create(generation.clone(), cli_id)?;
-    let anonymous = ControlClient::config(&endpoint.publication, None)?;
+    let anonymous = ControlClient::config_v2(&endpoint.publication, None)?;
     let mut enrollment =
-        ControlClient::connect(&endpoint.socket_path, anonymous, &endpoint.publication)?;
+        ControlClient::connect_v2(&endpoint.socket_path, anonymous, &endpoint.publication)?;
     ControlClient::send(
         &mut enrollment,
         &ControlRequest::Enroll(pending.request().clone()),
@@ -442,12 +444,12 @@ pub fn execute_interactive_over_control_face_as<E, H: HostTerminal>(
         session_id,
         session_anchor_public_key: identity.anchor.verifying_key().to_bytes().to_vec(),
     };
-    let authenticated = ControlClient::config(
+    let authenticated = ControlClient::config_v2(
         &endpoint.publication,
         Some((&credential, pending.private_key())),
     )?;
     let mut stream =
-        ControlClient::connect(&endpoint.socket_path, authenticated, &endpoint.publication)?;
+        ControlClient::connect_v2(&endpoint.socket_path, authenticated, &endpoint.publication)?;
     ControlClient::send(
         &mut stream,
         &ControlRequest::Register(SessionRegistrationRequest {
@@ -466,13 +468,70 @@ pub fn execute_interactive_over_control_face_as<E, H: HostTerminal>(
         ));
     }
     endpoint.acknowledge_waiter_transfer()?;
-    let mut transport = ControlFaceInteractiveTransport { stream };
+    let mut transport = ControlFaceInteractiveTransport::new(stream).map_err(widen_client_error)?;
     run_interactive_session(&mut transport, terminal, identity, &binding, argv)
         .map_err(widen_client_error)
 }
 
 struct ControlFaceInteractiveTransport {
-    stream: rustls::StreamOwned<rustls::ClientConnection, std::os::unix::net::UnixStream>,
+    commands: Option<mpsc::Sender<InteractiveTransportCommand>>,
+    batches:
+        mpsc::Receiver<Result<InteractiveSessionBatchV2, ClientError<std::convert::Infallible>>>,
+    cancellation: std::os::unix::net::UnixStream,
+    worker: Option<JoinHandle<()>>,
+}
+
+enum InteractiveTransportCommand {
+    Start(InteractiveSessionRequestV2),
+    Event(InteractiveSessionEventV2),
+}
+
+impl ControlFaceInteractiveTransport {
+    fn new(
+        mut stream: rustls::StreamOwned<rustls::ClientConnection, std::os::unix::net::UnixStream>,
+    ) -> Result<Self, ClientError<std::convert::Infallible>> {
+        let cancellation = stream.sock.try_clone()?;
+        let (commands, command_rx) = mpsc::channel();
+        let (batch_tx, batches) = mpsc::channel();
+        let worker = std::thread::spawn(move || {
+            while let Ok(command) = command_rx.recv() {
+                let result = match command {
+                    InteractiveTransportCommand::Start(request) => ControlClient::send(
+                        &mut stream,
+                        &ControlRequest::ExecuteInteractiveV2(request),
+                    )
+                    .map_err(map_control_connection)
+                    .and_then(|()| receive_interactive_batch(&mut stream)),
+                    InteractiveTransportCommand::Event(event) => {
+                        ControlClient::send(&mut stream, &event)
+                            .map_err(map_control_connection)
+                            .and_then(|()| receive_interactive_batch(&mut stream))
+                    }
+                };
+                let failed = result.is_err();
+                if batch_tx.send(result).is_err() || failed {
+                    break;
+                }
+            }
+        });
+        Ok(Self {
+            commands: Some(commands),
+            batches,
+            cancellation,
+            worker: Some(worker),
+        })
+    }
+
+    fn submit(
+        &self,
+        command: InteractiveTransportCommand,
+    ) -> Result<(), ClientError<std::convert::Infallible>> {
+        self.commands
+            .as_ref()
+            .ok_or(ClientError::LostConnection)?
+            .send(command)
+            .map_err(|_| ClientError::LostConnection)
+    }
 }
 
 impl InteractiveSessionTransport for ControlFaceInteractiveTransport {
@@ -480,27 +539,43 @@ impl InteractiveSessionTransport for ControlFaceInteractiveTransport {
 
     fn start(
         &mut self,
-        request: InteractiveSessionRequest,
-    ) -> Result<InteractiveSessionBatch, ClientError<Self::Error>> {
-        ControlClient::send(
-            &mut self.stream,
-            &ControlRequest::ExecuteInteractive(request),
-        )?;
-        receive_interactive_batch(&mut self.stream)
+        request: InteractiveSessionRequestV2,
+    ) -> Result<(), ClientError<Self::Error>> {
+        self.submit(InteractiveTransportCommand::Start(request))
     }
 
-    fn exchange(
+    fn send(&mut self, event: InteractiveSessionEventV2) -> Result<(), ClientError<Self::Error>> {
+        self.submit(InteractiveTransportCommand::Event(event))
+    }
+
+    fn try_receive(
         &mut self,
-        event: InteractiveSessionEvent,
-    ) -> Result<InteractiveSessionBatch, ClientError<Self::Error>> {
-        ControlClient::send(&mut self.stream, &event).map_err(map_control_connection)?;
-        receive_interactive_batch(&mut self.stream)
+    ) -> Result<Option<InteractiveSessionBatchV2>, ClientError<Self::Error>> {
+        match self.batches.try_recv() {
+            Ok(batch) => batch.map(Some),
+            Err(mpsc::TryRecvError::Empty) => Ok(None),
+            Err(mpsc::TryRecvError::Disconnected) => Err(ClientError::LostConnection),
+        }
+    }
+
+    fn cancel(&mut self) {
+        self.commands.take();
+        let _ = self.cancellation.shutdown(Shutdown::Both);
+    }
+}
+
+impl Drop for ControlFaceInteractiveTransport {
+    fn drop(&mut self) {
+        self.cancel();
+        if let Some(worker) = self.worker.take() {
+            let _ = worker.join();
+        }
     }
 }
 
 fn receive_interactive_batch(
     stream: &mut impl std::io::Read,
-) -> Result<InteractiveSessionBatch, ClientError<std::convert::Infallible>> {
+) -> Result<InteractiveSessionBatchV2, ClientError<std::convert::Infallible>> {
     ControlClient::receive(stream).map_err(map_control_connection)
 }