Unify client Domain observation

crates/lazaret-client/src/coordinator_backend.rs

@@ -6,7 +6,6 @@ use std::io::{self, IsTerminal, Write};
 use std::os::fd::{AsFd, AsRawFd};
 use std::os::unix::fs::MetadataExt;
 use std::path::PathBuf;
-use std::thread;
 use std::time::Duration;
 
 use lazaret_core::configuration::{
@@ -25,6 +24,10 @@ use lazaret_runtime_supervisor::sandbox::default_sandbox_layout;
 use rustix::fs::{Mode, OFlags, openat};
 use rustix::pipe::{PipeFlags, pipe_with};
 
+use crate::domain_observation::{
+    DiscoveryAction, DiscoveryState, DomainObservation, ObservationPath, SystemObservationClock,
+    WaiterAction,
+};
 use crate::{
     BootstrapAcquisition, ClientBackend, ClientCommand, ClientError, ClientSessionIdentity,
     ControlEndpoint, DomainStatus, execute_interactive_over_control_face_as,
@@ -176,22 +179,60 @@ impl CoordinatorClientBackend {
         }
     }
 
-    fn await_publication(
+    fn observe_domain(
         &mut self,
+        initial_path: ObservationPath,
     ) -> Result<ControlEndpoint, ClientError<CoordinatorClientError>> {
+        let waiter = if initial_path == ObservationPath::WaitingClient {
+            let configuration = self.configuration().map_err(ClientError::Backend)?;
+            let runtime_authority = configuration
+                .runtime_domain_root()
+                .open_authority()
+                .map_err(failure)
+                .map_err(ClientError::Backend)?;
+            let socket = PathBuf::from(format!(
+                "/proc/self/fd/{}/{}",
+                runtime_authority.as_raw_fd(),
+                BOOTSTRAP_WAITER_SOCKET_NAME
+            ));
+            let request = BootstrapWaitRequest {
+                protocol_version: lazaret_core::records::ProtocolVersion::V1,
+                domain_id: configuration.domain_id(),
+                configuration_fingerprint: configuration.configuration_fingerprint(),
+                cli_id: self.identity.cli_id(),
+                session_id: self.identity.session_id(),
+                session_anchor_id: self.identity.session_anchor_id(),
+            };
+            Some((runtime_authority, socket, request))
+        } else {
+            None
+        };
+        let mut observer = DomainObservation::new(initial_path, SystemObservationClock::start());
+        let mut waiter_transfer = None;
         loop {
-            match self.discover().map_err(ClientError::Backend)? {
+            let (state, endpoint) = match self.discover().map_err(ClientError::Backend)? {
-                Discovery::Ready(endpoint) => return Ok(*endpoint),
+                Discovery::Ready(endpoint) => (DiscoveryState::Ready, Some(*endpoint)),
-                Discovery::Stale | Discovery::Absent => {}
+                Discovery::Stale => (DiscoveryState::Stale, None),
+                Discovery::Absent => (DiscoveryState::Absent, None),
+            };
+            match observer.discovery(state) {
+                DiscoveryAction::ReturnReady => {
+                    let endpoint = endpoint.expect("ready discovery supplies an endpoint");
+                    return Ok(match waiter_transfer {
+                        Some(stream) => endpoint.retain_waiter_transfer(stream),
+                        None => endpoint,
+                    });
                 }
-            if self
+                DiscoveryAction::RefuseStale => return Err(ClientError::StaleCoordinator),
+                DiscoveryAction::CheckCoordinator => {
+                    let exited = self
                         .coordinator
                         .as_ref()
                         .ok_or_else(|| ClientError::Backend(failure("Host Coordinator is absent")))?
                         .has_exited()
                         .map_err(failure)
-                .map_err(ClientError::Backend)?
+                        .map_err(ClientError::Backend)?;
-            {
+                    if observer.coordinator_exited(exited) {
                         let status = self
                             .coordinator
                             .take()
@@ -203,7 +244,95 @@ impl CoordinatorClientBackend {
                             "Host Coordinator exited before publication with {status:?}"
                         ))));
                     }
-            thread::sleep(Duration::from_millis(10));
+                }
+                DiscoveryAction::AttemptWaiter => {
+                    let (_, socket, request) = waiter
+                        .as_ref()
+                        .expect("waiting observation retains waiter authority");
+                    if let Some(stream) = self.attempt_waiter(&observer, socket, request)? {
+                        waiter_transfer = Some(stream);
+                        observer.transfer_publication();
+                        continue;
+                    }
+                }
+                DiscoveryAction::Retry => {}
+            }
+            observer
+                .retry()
+                .map_err(|_| observation_deadline(observer.path()))?;
+        }
+    }
+
+    fn attempt_waiter(
+        &self,
+        observer: &DomainObservation<SystemObservationClock>,
+        socket: &std::path::Path,
+        request: &BootstrapWaitRequest,
+    ) -> Result<Option<std::os::unix::net::UnixStream>, ClientError<CoordinatorClientError>> {
+        let mut stream = match std::os::unix::net::UnixStream::connect(socket) {
+            Ok(stream) => stream,
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    io::ErrorKind::NotFound | io::ErrorKind::ConnectionRefused
+                ) =>
+            {
+                if observer.lock_available(
+                    self.domain_lock_is_available()
+                        .map_err(ClientError::Backend)?,
+                ) == Some(WaiterAction::RefuseFailed)
+                {
+                    return Err(ClientError::WaitingDomainFailed);
+                }
+                return Ok(None);
+            }
+            Err(error) => return Err(ClientError::Backend(failure(error))),
+        };
+        stream
+            .set_read_timeout(Some(observation_remaining(observer)?))
+            .map_err(failure)
+            .map_err(ClientError::Backend)?;
+        write_frame(&mut stream, request).map_err(|error| {
+            ClientError::Backend(failure(format!("bootstrap waiter request failed: {error}")))
+        })?;
+        let first: BootstrapWaitOutcome = read_frame(&mut stream).map_err(|error| {
+            ClientError::Backend(failure(format!(
+                "bootstrap waiter admission failed: {error}"
+            )))
+        })?;
+        let mut action = observer.waiter(first, false);
+        if action == WaiterAction::AwaitTerminal {
+            action = match read_frame(&mut stream) {
+                Ok(outcome) => observer.waiter(outcome, true),
+                Err(_) => WaiterAction::RefuseFailed,
+            };
+        }
+        match action {
+            WaiterAction::AwaitPublication => Ok(Some(stream)),
+            WaiterAction::RefuseStale => Err(ClientError::StaleCoordinator),
+            WaiterAction::RefuseFailed => Err(ClientError::WaitingDomainFailed),
+            WaiterAction::RefuseProtocol | WaiterAction::AwaitTerminal => Err(
+                ClientError::Protocol("bootstrap waiter sent an invalid terminal outcome"),
+            ),
+        }
+    }
+}
+
+fn observation_remaining(
+    observer: &DomainObservation<SystemObservationClock>,
+) -> Result<Duration, ClientError<CoordinatorClientError>> {
+    observer
+        .remaining()
+        .map_err(|_| observation_deadline(observer.path()))
+}
+
+fn observation_deadline(path: ObservationPath) -> ClientError<CoordinatorClientError> {
+    match path {
+        ObservationPath::BootstrapLeader => {
+            ClientError::Backend(failure("Host Coordinator publication deadline expired"))
+        }
+        ObservationPath::WaitingClient | ObservationPath::PublicationTransfer => {
+            ClientError::WaitingDomainFailed
         }
     }
 }
@@ -247,85 +376,11 @@ impl ClientBackend for CoordinatorClientBackend {
     }
 
     fn bootstrap(&mut self) -> Result<Self::Endpoint, ClientError<Self::Error>> {
-        self.await_publication()
+        self.observe_domain(ObservationPath::BootstrapLeader)
     }
 
     fn wait_until_ready(&mut self) -> Result<Self::Endpoint, ClientError<Self::Error>> {
-        let configuration = self.configuration().map_err(ClientError::Backend)?;
+        self.observe_domain(ObservationPath::WaitingClient)
-        let runtime_authority = configuration
-            .runtime_domain_root()
-            .open_authority()
-            .map_err(failure)
-            .map_err(ClientError::Backend)?;
-        let socket = PathBuf::from(format!(
-            "/proc/self/fd/{}/{}",
-            runtime_authority.as_raw_fd(),
-            BOOTSTRAP_WAITER_SOCKET_NAME
-        ));
-        let request = BootstrapWaitRequest {
-            protocol_version: lazaret_core::records::ProtocolVersion::V1,
-            domain_id: configuration.domain_id(),
-            configuration_fingerprint: configuration.configuration_fingerprint(),
-            cli_id: self.identity.cli_id(),
-            session_id: self.identity.session_id(),
-            session_anchor_id: self.identity.session_anchor_id(),
-        };
-        loop {
-            match self.discover().map_err(ClientError::Backend)? {
-                Discovery::Ready(endpoint) => return Ok(*endpoint),
-                Discovery::Stale | Discovery::Absent => {}
-            }
-            match std::os::unix::net::UnixStream::connect(&socket) {
-                Ok(mut stream) => {
-                    write_frame(&mut stream, &request).map_err(|error| {
-                        ClientError::Backend(failure(format!(
-                            "bootstrap waiter request failed: {error}"
-                        )))
-                    })?;
-                    let first: BootstrapWaitOutcome = read_frame(&mut stream).map_err(|error| {
-                        ClientError::Backend(failure(format!(
-                            "bootstrap waiter admission failed: {error}"
-                        )))
-                    })?;
-                    if first != BootstrapWaitOutcome::Waiting {
-                        return match first {
-                            BootstrapWaitOutcome::Refused => Err(ClientError::StaleCoordinator),
-                            BootstrapWaitOutcome::Failed => Err(ClientError::WaitingDomainFailed),
-                            BootstrapWaitOutcome::Published => {
-                                self.await_ready_publication(Some(stream))
-                            }
-                            BootstrapWaitOutcome::Waiting => unreachable!(),
-                        };
-                    }
-                    return match read_frame(&mut stream) {
-                        Ok(BootstrapWaitOutcome::Published) => {
-                            self.await_ready_publication(Some(stream))
-                        }
-                        Ok(BootstrapWaitOutcome::Failed) | Err(_) => {
-                            Err(ClientError::WaitingDomainFailed)
-                        }
-                        Ok(BootstrapWaitOutcome::Waiting | BootstrapWaitOutcome::Refused) => {
-                            Err(ClientError::Protocol(
-                                "bootstrap waiter sent an invalid terminal outcome",
-                            ))
-                        }
-                    };
-                }
-                Err(error)
-                    if matches!(
-                        error.kind(),
-                        io::ErrorKind::NotFound | io::ErrorKind::ConnectionRefused
-                    ) => {}
-                Err(error) => return Err(ClientError::Backend(failure(error))),
-            }
-            if self
-                .domain_lock_is_available()
-                .map_err(ClientError::Backend)?
-            {
-                return Err(ClientError::WaitingDomainFailed);
-            }
-            thread::sleep(Duration::from_millis(10));
-        }
     }
 
     fn execute(
@@ -362,27 +417,6 @@ impl ClientBackend for CoordinatorClientBackend {
     }
 }
 
-impl CoordinatorClientBackend {
-    fn await_ready_publication(
-        &self,
-        waiter: Option<std::os::unix::net::UnixStream>,
-    ) -> Result<ControlEndpoint, ClientError<CoordinatorClientError>> {
-        loop {
-            match self.discover().map_err(ClientError::Backend)? {
-                Discovery::Ready(endpoint) => {
-                    let endpoint = *endpoint;
-                    return Ok(match waiter {
-                        Some(stream) => endpoint.retain_waiter_transfer(stream),
-                        None => endpoint,
-                    });
-                }
-                Discovery::Stale => return Err(ClientError::StaleCoordinator),
-                Discovery::Absent => thread::sleep(Duration::from_millis(1)),
-            }
-        }
-    }
-}
-
 pub(crate) fn open_private_domain_lock(root: &impl AsFd) -> Result<File, CoordinatorClientError> {
     let file = File::from(
         openat(

crates/lazaret-client/src/domain_observation.rs

@@ -0,0 +1,288 @@
+use std::time::{Duration, Instant};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum ObservationPath {
+    BootstrapLeader,
+    WaitingClient,
+    PublicationTransfer,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum DiscoveryState {
+    Ready,
+    Stale,
+    Absent,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum DiscoveryAction {
+    ReturnReady,
+    CheckCoordinator,
+    AttemptWaiter,
+    Retry,
+    RefuseStale,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum WaiterAction {
+    AwaitTerminal,
+    AwaitPublication,
+    RefuseStale,
+    RefuseFailed,
+    RefuseProtocol,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum ObservationFailure {
+    DeadlineExceeded,
+}
+
+#[derive(Clone, Copy)]
+struct ObservationPolicy {
+    retry: Duration,
+    deadline: Duration,
+}
+
+const BOOTSTRAP_POLICY: ObservationPolicy = ObservationPolicy {
+    retry: Duration::from_millis(10),
+    deadline: Duration::from_secs(30),
+};
+const WAITING_POLICY: ObservationPolicy = ObservationPolicy {
+    retry: Duration::from_millis(10),
+    deadline: Duration::from_secs(30),
+};
+const TRANSFER_POLICY: ObservationPolicy = ObservationPolicy {
+    retry: Duration::from_millis(1),
+    deadline: Duration::from_secs(30),
+};
+
+pub(crate) trait ObservationClock {
+    fn elapsed(&self) -> Duration;
+    fn wait(&mut self, duration: Duration);
+    fn restart(&mut self);
+}
+
+pub(crate) struct SystemObservationClock {
+    started: Instant,
+}
+
+impl SystemObservationClock {
+    pub(crate) fn start() -> Self {
+        Self {
+            started: Instant::now(),
+        }
+    }
+}
+
+impl ObservationClock for SystemObservationClock {
+    fn elapsed(&self) -> Duration {
+        self.started.elapsed()
+    }
+
+    fn wait(&mut self, duration: Duration) {
+        std::thread::sleep(duration);
+    }
+
+    fn restart(&mut self) {
+        self.started = Instant::now();
+    }
+}
+
+pub(crate) struct DomainObservation<C> {
+    path: ObservationPath,
+    clock: C,
+}
+
+impl<C: ObservationClock> DomainObservation<C> {
+    pub(crate) fn new(path: ObservationPath, clock: C) -> Self {
+        Self { path, clock }
+    }
+
+    pub(crate) fn path(&self) -> ObservationPath {
+        self.path
+    }
+
+    pub(crate) fn transfer_publication(&mut self) {
+        self.path = ObservationPath::PublicationTransfer;
+        self.clock.restart();
+    }
+
+    pub(crate) fn discovery(&self, state: DiscoveryState) -> DiscoveryAction {
+        match (self.path, state) {
+            (_, DiscoveryState::Ready) => DiscoveryAction::ReturnReady,
+            (ObservationPath::BootstrapLeader, _) => DiscoveryAction::CheckCoordinator,
+            (ObservationPath::WaitingClient, _) => DiscoveryAction::AttemptWaiter,
+            (ObservationPath::PublicationTransfer, DiscoveryState::Stale) => {
+                DiscoveryAction::RefuseStale
+            }
+            (ObservationPath::PublicationTransfer, DiscoveryState::Absent) => {
+                DiscoveryAction::Retry
+            }
+        }
+    }
+
+    pub(crate) fn waiter(
+        &self,
+        outcome: lazaret_host_coordinator::waiter::BootstrapWaitOutcome,
+        admitted: bool,
+    ) -> WaiterAction {
+        use lazaret_host_coordinator::waiter::BootstrapWaitOutcome::{
+            Failed, Published, Refused, Waiting,
+        };
+        if self.path != ObservationPath::WaitingClient {
+            return WaiterAction::RefuseProtocol;
+        }
+        match (admitted, outcome) {
+            (false, Waiting) => WaiterAction::AwaitTerminal,
+            (_, Published) => WaiterAction::AwaitPublication,
+            (false, Refused) => WaiterAction::RefuseStale,
+            (_, Failed) => WaiterAction::RefuseFailed,
+            _ => WaiterAction::RefuseProtocol,
+        }
+    }
+
+    pub(crate) fn lock_available(&self, available: bool) -> Option<WaiterAction> {
+        (self.path == ObservationPath::WaitingClient && available)
+            .then_some(WaiterAction::RefuseFailed)
+    }
+
+    pub(crate) fn coordinator_exited(&self, exited: bool) -> bool {
+        self.path == ObservationPath::BootstrapLeader && exited
+    }
+
+    pub(crate) fn remaining(&self) -> Result<Duration, ObservationFailure> {
+        let remaining = self.policy().deadline.saturating_sub(self.clock.elapsed());
+        if remaining.is_zero() {
+            Err(ObservationFailure::DeadlineExceeded)
+        } else {
+            Ok(remaining)
+        }
+    }
+
+    pub(crate) fn retry(&mut self) -> Result<(), ObservationFailure> {
+        let remaining = self.remaining()?;
+        self.clock.wait(self.policy().retry.min(remaining));
+        self.remaining().map(|_| ())
+    }
+
+    fn policy(&self) -> ObservationPolicy {
+        match self.path {
+            ObservationPath::BootstrapLeader => BOOTSTRAP_POLICY,
+            ObservationPath::WaitingClient => WAITING_POLICY,
+            ObservationPath::PublicationTransfer => TRANSFER_POLICY,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use lazaret_host_coordinator::waiter::BootstrapWaitOutcome;
+
+    #[derive(Default)]
+    struct FakeClock {
+        elapsed: Duration,
+        waits: Vec<Duration>,
+    }
+
+    impl ObservationClock for FakeClock {
+        fn elapsed(&self) -> Duration {
+            self.elapsed
+        }
+
+        fn wait(&mut self, duration: Duration) {
+            self.elapsed += duration;
+            self.waits.push(duration);
+        }
+
+        fn restart(&mut self) {
+            self.elapsed = Duration::ZERO;
+        }
+    }
+
+    #[test]
+    fn every_path_has_explicit_discovery_actions() {
+        let cases = [
+            (
+                ObservationPath::BootstrapLeader,
+                [
+                    DiscoveryAction::ReturnReady,
+                    DiscoveryAction::CheckCoordinator,
+                    DiscoveryAction::CheckCoordinator,
+                ],
+            ),
+            (
+                ObservationPath::WaitingClient,
+                [
+                    DiscoveryAction::ReturnReady,
+                    DiscoveryAction::AttemptWaiter,
+                    DiscoveryAction::AttemptWaiter,
+                ],
+            ),
+            (
+                ObservationPath::PublicationTransfer,
+                [
+                    DiscoveryAction::ReturnReady,
+                    DiscoveryAction::RefuseStale,
+                    DiscoveryAction::Retry,
+                ],
+            ),
+        ];
+        for (path, expected) in cases {
+            let observer = DomainObservation::new(path, FakeClock::default());
+            assert_eq!(observer.discovery(DiscoveryState::Ready), expected[0]);
+            assert_eq!(observer.discovery(DiscoveryState::Stale), expected[1]);
+            assert_eq!(observer.discovery(DiscoveryState::Absent), expected[2]);
+        }
+    }
+
+    #[test]
+    fn waiter_lock_and_coordinator_outcomes_are_closed() {
+        let observer = DomainObservation::new(ObservationPath::WaitingClient, FakeClock::default());
+        assert_eq!(
+            observer.waiter(BootstrapWaitOutcome::Waiting, false),
+            WaiterAction::AwaitTerminal
+        );
+        assert_eq!(
+            observer.waiter(BootstrapWaitOutcome::Published, false),
+            WaiterAction::AwaitPublication
+        );
+        assert_eq!(
+            observer.waiter(BootstrapWaitOutcome::Failed, true),
+            WaiterAction::RefuseFailed
+        );
+        assert_eq!(
+            observer.waiter(BootstrapWaitOutcome::Refused, false),
+            WaiterAction::RefuseStale
+        );
+        assert_eq!(
+            observer.waiter(BootstrapWaitOutcome::Waiting, true),
+            WaiterAction::RefuseProtocol
+        );
+        assert_eq!(
+            observer.lock_available(true),
+            Some(WaiterAction::RefuseFailed)
+        );
+        assert_eq!(observer.lock_available(false), None);
+
+        let leader = DomainObservation::new(ObservationPath::BootstrapLeader, FakeClock::default());
+        assert!(leader.coordinator_exited(true));
+        assert!(!leader.coordinator_exited(false));
+        assert!(!observer.coordinator_exited(true));
+    }
+
+    #[test]
+    fn retry_deadlines_and_transfer_restart_are_deterministic() {
+        let mut observer =
+            DomainObservation::new(ObservationPath::WaitingClient, FakeClock::default());
+        observer.retry().unwrap();
+        assert_eq!(observer.clock.waits, [Duration::from_millis(10)]);
+        observer.clock.elapsed = WAITING_POLICY.deadline;
+        assert_eq!(observer.retry(), Err(ObservationFailure::DeadlineExceeded));
+        observer.transfer_publication();
+        assert_eq!(observer.path(), ObservationPath::PublicationTransfer);
+        assert_eq!(observer.clock.elapsed, Duration::ZERO);
+        observer.retry().unwrap();
+        assert_eq!(observer.clock.waits.last(), Some(&Duration::from_millis(1)));
+    }
+}

crates/lazaret-client/src/lib.rs

@@ -22,6 +22,7 @@ use crate::terminal::{HostTerminal, InteractiveSessionTransport, run_interactive
 
 pub mod cli;
 pub mod coordinator_backend;
+mod domain_observation;
 pub mod state;
 pub mod terminal;