1 files changed, 59 insertions, 0 deletions
diff --git a/src/client/endpoint_commands.rs b/src/client/endpoint_commands.rs
index decb8522..d07b3d42 100644
--- a/src/client/endpoint_commands.rs
+++ b/src/client/endpoint_commands.rs
@@ -1,5 +1,6 @@
 use std::collections::VecDeque;
 use std::io;
+use std::time::{Duration, Instant};
 
 use crate::api::client::ApiClientError;
 use crate::api::schema::{Request, ResponseResult};
@@ -8,10 +9,14 @@ use crate::protocol::ClientMessage;
 
 use super::shell::ClientShellEndpointError;
 
+const ENDPOINT_COMMAND_TIMEOUT: Duration = Duration::from_secs(60);
+
 struct InFlightCommand {
     boot_id: String,
     request_id: String,
     response: Vec<u8>,
+    sent_at: Instant,
+    timed_out: bool,
 }
 
 pub(super) struct EndpointCommandResult {
@@ -52,10 +57,30 @@ impl EndpointCommands {
             boot_id,
             request_id,
             response: Vec::new(),
+            sent_at: Instant::now(),
+            timed_out: false,
         });
         Ok(())
     }
 
+    pub(super) fn expire(&mut self, now: Instant) -> Option<EndpointCommandResult> {
+        let command = self.in_flight.as_mut()?;
+        if command.timed_out
+            || now.saturating_duration_since(command.sent_at) < ENDPOINT_COMMAND_TIMEOUT
+        {
+            return None;
+        }
+        command.timed_out = true;
+        Some(EndpointCommandResult {
+            boot_id: command.boot_id.clone(),
+            request_id: command.request_id.clone(),
+            result: Err(ClientShellEndpointError {
+                code: Some("endpoint_timeout".into()),
+                message: "this server did not respond to the action".into(),
+            }),
+        })
+    }
+
     pub(super) fn receive_chunk(
         &mut self,
         response_boot_id: &str,
@@ -140,6 +165,8 @@ mod tests {
                 boot_id: "boot-a".into(),
                 request_id: "request-a".into(),
                 response: Vec::new(),
+                sent_at: Instant::now(),
+                timed_out: false,
             }),
             ..EndpointCommands::default()
         }
@@ -212,6 +239,38 @@ mod tests {
     }
 
     #[test]
+    fn in_flight_endpoint_command_expires_and_releases_the_lane() {
+        let mut commands = commands_with_in_flight();
+        let expired = commands
+            .expire(std::time::Instant::now() + ENDPOINT_COMMAND_TIMEOUT)
+            .expect("expired endpoint command");
+
+        assert_eq!(expired.boot_id, "boot-a");
+        assert_eq!(expired.request_id, "request-a");
+        assert!(matches!(
+            expired.result,
+            Err(ClientShellEndpointError {
+                code: Some(code),
+                ..
+            }) if code == "endpoint_timeout"
+        ));
+        assert!(commands.in_flight.is_some());
+        assert!(commands
+            .expire(std::time::Instant::now() + ENDPOINT_COMMAND_TIMEOUT)
+            .is_none());
+        let late_response = serde_json::to_vec(&SuccessResponse {
+            id: "request-a".into(),
+            result: ResponseResult::Ok {},
+        })
+        .unwrap();
+        assert!(commands
+            .receive_chunk("boot-a", "request-a", true, late_response)
+            .expect("late response releases the synchronized lane")
+            .is_some());
+        assert!(commands.in_flight.is_none());
+    }
+
+    #[test]
     fn response_from_another_boot_is_rejected() {
         let mut commands = commands_with_in_flight();