diff options
Diffstat (limited to 'src/client/endpoint')
| -rw-r--r-- | src/client/endpoint/activation.rs | 1152 | ||||
| -rw-r--r-- | src/client/endpoint/activation/model.rs | 174 | ||||
| -rw-r--r-- | src/client/endpoint/activation/protocol.rs | 219 | ||||
| -rw-r--r-- | src/client/endpoint/activation_tests.rs | 1305 | ||||
| -rw-r--r-- | src/client/endpoint/catalog.rs | 578 | ||||
| -rw-r--r-- | src/client/endpoint/control.rs | 63 | ||||
| -rw-r--r-- | src/client/endpoint/health.rs | 99 | ||||
| -rw-r--r-- | src/client/endpoint/message_policy.rs | 124 | ||||
| -rw-r--r-- | src/client/endpoint/registry.rs | 629 | ||||
| -rw-r--r-- | src/client/endpoint/supervisor.rs | 492 | ||||
| -rw-r--r-- | src/client/endpoint/writer.rs | 400 |
11 files changed, 5235 insertions, 0 deletions
diff --git a/src/client/endpoint/activation.rs b/src/client/endpoint/activation.rs new file mode 100644 index 00000000..a5b63344 --- /dev/null +++ b/src/client/endpoint/activation.rs @@ -0,0 +1,1152 @@ +use std::time::{Duration, Instant}; + +use super::{ClientEndpointId, ClientEndpointStatus, EndpointRegistry, EndpointSendOutcome}; + +mod model; +mod protocol; +pub(crate) use model::{ + ActivationBeginError, ActivationCompletion, ActivationRollback, EndpointActivationIntent, + PendingEndpointActivation, SurfaceActivationProgress, +}; +use model::{ActivationEvidence, ActivationPhase, EndpointLease}; +use protocol::*; + +const ACTIVATION_TIMEOUT: Duration = Duration::from_secs(5); + +impl PendingEndpointActivation { + #[allow(clippy::too_many_arguments)] + pub(crate) fn begin( + shell: &crate::client::shell::ClientShellState, + endpoints: &mut EndpointRegistry, + target: ClientEndpointId, + focus: Option<crate::client::shell::ClientEndpointFocusTarget>, + resize: crate::protocol::ClientMessage, + serial: u64, + now: Instant, + ) -> Result<Self, ActivationBeginError> { + resize_geometry(&resize).ok_or_else(|| { + ActivationBeginError::Preflight( + "endpoint activation did not include a surface resize".to_owned(), + ) + })?; + let source_id = endpoints.active_id().clone(); + let source_has_live_surface = endpoints + .connection(&source_id) + .is_some_and(|connection| connection.surface_active); + let (source, source_available) = if source_has_live_surface { + ( + endpoint_lease(shell, endpoints, &source_id) + .map_err(ActivationBeginError::Preflight)?, + true, + ) + } else { + (disconnected_endpoint_lease(shell, &source_id), false) + }; + let target_lease = + endpoint_lease(shell, endpoints, &target).map_err(ActivationBeginError::Preflight)?; + let source_compatible = !source_available + || endpoints + .connection(&source_id) + .is_some_and(|connection| connection.negotiation.supports_surface_interest()); + let target_compatible = endpoints + .connection(&target) + .is_some_and(|connection| connection.negotiation.supports_surface_interest()); + if !source_compatible || !target_compatible { + return Err(ActivationBeginError::Preflight( + "endpoint must be updated before it can join the selected surface".into(), + )); + } + + let source_is_target = source.endpoint_id == target_lease.endpoint_id; + // Validate every typed lifecycle and optional focus envelope before the first transport + // write. Any error above this line is guaranteed not to have changed either endpoint. + let source_release_request = (source_available && !source_is_target) + .then(|| { + surface_interest_request( + &source.boot_id, + format!("client-shell-surface:{serial}:off"), + false, + ) + }) + .transpose() + .map_err(|error| ActivationBeginError::Preflight(error.to_string()))?; + surface_interest_request( + &target_lease.boot_id, + format!("client-shell-surface:{serial}:on"), + true, + ) + .map_err(|error| ActivationBeginError::Preflight(error.to_string()))?; + if let Some(target) = focus.as_ref() { + focus_request( + &target_lease.boot_id, + format!("client-shell-focus:{serial}:1"), + target, + ) + .map_err(|error| ActivationBeginError::Preflight(error.to_string()))?; + } + + endpoints.freeze_input(); + let mut activation = Self { + source, + source_available, + target: target_lease, + focus, + host_focused: shell.host_focus_baseline(), + resize: resize.clone(), + phase: ActivationPhase::ReleasingSource { + request_id: format!("client-shell-surface:{serial}:off"), + }, + deadline: now + ACTIVATION_TIMEOUT, + epoch: serial, + next_focus_serial: 0, + rollback_error: None, + successor: None, + }; + + // Reconnecting the selected endpoint has no live source surface to release. All normal + // handoffs must make the source locally inactive before a target request is even sent. + if source_is_target || !source_available { + if let Err(error) = activation.start_target(endpoints, resize) { + return Err(ActivationBeginError::Partial { + activation: Box::new(activation), + error, + }); + } + } else { + // `surface.set(false)` removes the viewer, but old servers only emit the PTY focus + // loss while the viewer is still active. Revoke it explicitly before source-off. + if endpoints.send_to( + &activation.source.endpoint_id, + &crate::protocol::ClientMessage::ClientShellFocus { focused: false }, + ) != EndpointSendOutcome::Sent + { + return Err(ActivationBeginError::Partial { + activation: Box::new(activation), + error: "source endpoint focus revoke could not be sent".into(), + }); + } + let request = source_release_request.expect("validated source release request"); + if endpoints.send_to(&activation.source.endpoint_id, &request) + != EndpointSendOutcome::Sent + { + return Err(ActivationBeginError::Partial { + activation: Box::new(activation), + error: "source endpoint release could not be sent".into(), + }); + } + // This is deliberately before the acknowledgement: the source is no longer viewed + // locally while its release is in flight, and pane input is consequently blocked. + endpoints.set_surface_active(&activation.source.endpoint_id, false); + } + Ok(activation) + } + + pub(crate) fn target(&self) -> &ClientEndpointId { + &self.target.endpoint_id + } + + fn geometry(&self) -> crate::protocol::ClientSurfaceSize { + resize_geometry(&self.resize).expect("activation resize was validated before construction") + } + + pub(crate) fn presentation_sync_endpoint(&self) -> Option<&ClientEndpointId> { + match self.phase { + ActivationPhase::ActivatingTarget { .. } => Some(&self.target.endpoint_id), + ActivationPhase::RestoringSource { .. } => Some(&self.source.endpoint_id), + _ => None, + } + } + + /// The complete source command lane cannot safely cross source-off into a later presentation + /// epoch. Other endpoint lanes are not part of this retirement. + pub(crate) fn source_command_lane(&self) -> Option<&ClientEndpointId> { + (self.source_available && self.source.endpoint_id != self.target.endpoint_id) + .then_some(&self.source.endpoint_id) + } + + pub(crate) fn can_retarget(&self, endpoint_id: &ClientEndpointId) -> bool { + self.target.endpoint_id == *endpoint_id + && self.successor.is_none() + && matches!( + self.phase, + ActivationPhase::ReleasingSource { .. } | ActivationPhase::ActivatingTarget { .. } + ) + } + + /// Replace an in-flight handoff with the latest endpoint-qualified intent. The current + /// transaction is still reversed through target-off/source-on; the replacement is launched + /// by the caller only after the source's coherent restoration commits. + pub(crate) fn supersede( + &mut self, + endpoint_id: ClientEndpointId, + target: Option<crate::client::shell::ClientEndpointFocusTarget>, + endpoints: &mut EndpointRegistry, + ) -> ActivationRollback { + self.successor = Some(EndpointActivationIntent { + endpoint_id, + target, + }); + // Source-on is already ordered and must finish before any replacement is allowed to + // begin. Later rapid selections only replace the retained intent; they never turn a + // safe restoration into an unavailable state. + let source_restoration_in_flight = matches!( + self.phase, + ActivationPhase::RestoringSource { .. } + ) || matches!( + &self.phase, + ActivationPhase::SynchronizingPresentation { completion, .. } + if matches!(completion.as_ref(), ActivationCompletion::RestoredSource { .. }) + ); + if source_restoration_in_flight { + return ActivationRollback::Pending; + } + self.rollback( + endpoints, + "endpoint handoff superseded by a newer selection".into(), + false, + ) + } + + pub(crate) fn accepts_endpoint(&self, endpoint_id: &ClientEndpointId, generation: u64) -> bool { + (self.source_available + && self.source.endpoint_id == *endpoint_id + && self.source.generation == generation) + || (self.target.endpoint_id == *endpoint_id && self.target.generation == generation) + } + + pub(crate) fn involves_endpoint(&self, endpoint_id: &ClientEndpointId) -> bool { + self.source.endpoint_id == *endpoint_id || self.target.endpoint_id == *endpoint_id + } + + pub(crate) fn accepts_response( + &self, + endpoint_id: &ClientEndpointId, + generation: u64, + boot_id: &str, + request_id: &str, + ) -> bool { + match &self.phase { + ActivationPhase::ReleasingSource { + request_id: expected, + } => { + endpoint_matches(&self.source, endpoint_id, generation, boot_id) + && expected == request_id + } + ActivationPhase::ActivatingTarget { + request_id: expected, + focus_request_id, + .. + } => { + endpoint_matches(&self.target, endpoint_id, generation, boot_id) + && (expected == request_id || focus_request_id.as_deref() == Some(request_id)) + } + ActivationPhase::ReleasingTargetForRollback { + request_id: expected, + } => { + endpoint_matches(&self.target, endpoint_id, generation, boot_id) + && expected == request_id + } + ActivationPhase::RestoringSource { + request_id: expected, + .. + } => { + endpoint_matches(&self.source, endpoint_id, generation, boot_id) + && expected == request_id + } + ActivationPhase::SynchronizingPresentation { + lease, + request_id: expected, + .. + } => { + endpoint_matches(lease, endpoint_id, generation, boot_id) && expected == request_id + } + ActivationPhase::AwaitingPresentationEffects { .. } => false, + } + } + + pub(crate) fn expired(&self, now: Instant) -> bool { + now >= self.deadline + } + + #[cfg(test)] + pub(crate) fn receive_response( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + request_id: &str, + data: &[u8], + endpoints: &mut EndpointRegistry, + ) -> SurfaceActivationProgress { + let boot_id = if self.source.endpoint_id == *endpoint_id { + self.source.boot_id.clone() + } else { + self.target.boot_id.clone() + }; + self.receive_response_for_boot( + endpoint_id, + generation, + &boot_id, + request_id, + data, + endpoints, + ) + } + + pub(crate) fn receive_response_for_boot( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + boot_id: &str, + request_id: &str, + data: &[u8], + endpoints: &mut EndpointRegistry, + ) -> SurfaceActivationProgress { + if !self.accepts_response(endpoint_id, generation, boot_id, request_id) { + return SurfaceActivationProgress::Stale; + } + let result = match decode_endpoint_response(request_id, data) { + Ok(result) => result, + Err(error) => { + return SurfaceActivationProgress::Rejected { + source_release_rejected: matches!( + self.phase, + ActivationPhase::ReleasingSource { .. } + ) && error.code.is_some(), + message: error.message, + }; + } + }; + match &mut self.phase { + ActivationPhase::ReleasingSource { .. } => { + if let Err(message) = surface_set_revision(&result, false) { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + if let Err(message) = self.start_target(endpoints, self.resize.clone()) { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + SurfaceActivationProgress::Pending + } + ActivationPhase::ActivatingTarget { + request_id: surface_request_id, + acknowledged_revision, + .. + } if surface_request_id == request_id => { + let revision = match surface_set_revision(&result, true) { + Ok(revision) => revision, + Err(message) => { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + }; + *acknowledged_revision = Some(revision); + self.progress() + } + ActivationPhase::ActivatingTarget { + focus_request_id, + focus_request_target, + focus_acknowledged, + .. + } if focus_request_id.as_deref() == Some(request_id) => { + let requested = focus_request_target.clone(); + let Some(requested) = requested else { + return SurfaceActivationProgress::Stale; + }; + if !focus_result_matches(Some(&requested), &result) { + return SurfaceActivationProgress::Rejected { + message: "endpoint focus returned an unexpected result".into(), + source_release_rejected: false, + }; + } + *focus_request_id = None; + *focus_request_target = None; + if self.focus != Some(requested) { + *focus_acknowledged = false; + if let Err(message) = self.send_latest_focus(endpoints) { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + return self.progress(); + } + *focus_acknowledged = true; + self.progress() + } + ActivationPhase::ReleasingTargetForRollback { .. } => { + if let Err(message) = surface_set_revision(&result, false) { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + endpoints.set_surface_active(&self.target.endpoint_id, false); + if !self.source_available { + return SurfaceActivationProgress::Rejected { + message: self.rollback_error.clone().unwrap_or_else(|| { + "the previous endpoint is no longer connected".into() + }), + source_release_rejected: false, + }; + } + if let Err(message) = self.start_source_restore(endpoints, self.resize.clone()) { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + SurfaceActivationProgress::Pending + } + ActivationPhase::RestoringSource { + acknowledged_revision, + .. + } => { + let revision = match surface_set_revision(&result, true) { + Ok(revision) => revision, + Err(message) => { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + }; + *acknowledged_revision = Some(revision); + self.progress() + } + ActivationPhase::SynchronizingPresentation { + acknowledged_revision, + .. + } => { + let revision = match surface_set_revision(&result, true) { + Ok(revision) => revision, + Err(message) => { + return SurfaceActivationProgress::Rejected { + message, + source_release_rejected: false, + }; + } + }; + *acknowledged_revision = Some(revision); + self.progress() + } + _ => SurfaceActivationProgress::Stale, + } + } + + pub(crate) fn receive_snapshot( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + snapshot: &crate::protocol::ClientShellSnapshot, + ) -> SurfaceActivationProgress { + let lease = match &self.phase { + ActivationPhase::ActivatingTarget { .. } => &self.target, + ActivationPhase::RestoringSource { .. } => &self.source, + ActivationPhase::SynchronizingPresentation { lease, .. } => lease, + _ => return SurfaceActivationProgress::Stale, + }; + if !endpoint_matches(lease, endpoint_id, generation, &snapshot.boot_id) + || snapshot.revision < lease.minimum_revision + { + return SurfaceActivationProgress::Stale; + } + let evidence = match &mut self.phase { + ActivationPhase::ActivatingTarget { evidence, .. } + if endpoint_matches(&self.target, endpoint_id, generation, &snapshot.boot_id) => + { + evidence + } + ActivationPhase::RestoringSource { evidence, .. } + if endpoint_matches(&self.source, endpoint_id, generation, &snapshot.boot_id) => + { + evidence + } + ActivationPhase::SynchronizingPresentation { + lease, evidence, .. + } if endpoint_matches(lease, endpoint_id, generation, &snapshot.boot_id) => evidence, + _ => return SurfaceActivationProgress::Stale, + }; + evidence.record_snapshot(snapshot); + self.progress() + } + + pub(crate) fn receive_surface( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + surface: crate::protocol::PaneSurfaceFrame, + ) -> SurfaceActivationProgress { + let lease = match &self.phase { + ActivationPhase::ActivatingTarget { .. } => &self.target, + ActivationPhase::RestoringSource { .. } => &self.source, + ActivationPhase::SynchronizingPresentation { lease, .. } => lease, + _ => return SurfaceActivationProgress::Stale, + }; + if !endpoint_matches(lease, endpoint_id, generation, &surface.boot_id) { + return SurfaceActivationProgress::Stale; + } + if !surface_matches_geometry(&surface, self.geometry()) { + return SurfaceActivationProgress::Pending; + } + match &mut self.phase { + ActivationPhase::ActivatingTarget { evidence, .. } + | ActivationPhase::RestoringSource { evidence, .. } + | ActivationPhase::SynchronizingPresentation { evidence, .. } => { + evidence.record_surface(surface) + } + _ => unreachable!("checked activation phase"), + } + self.progress() + } + + pub(crate) fn receive_presentation_effects_ready( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + token: &str, + ) -> SurfaceActivationProgress { + let ActivationPhase::AwaitingPresentationEffects { + lease, + token: expected, + ready, + .. + } = &mut self.phase + else { + return SurfaceActivationProgress::Stale; + }; + if lease.endpoint_id != *endpoint_id || lease.generation != generation || expected != token + { + return SurfaceActivationProgress::Stale; + } + *ready = true; + SurfaceActivationProgress::Ready + } + + /// A same-endpoint navigation request replaces the desired target but never joins the + /// in-flight focus RPC. Once that request resolves, `send_latest_focus` sends only the most + /// recent desired target. + pub(crate) fn retarget( + &mut self, + focus: Option<crate::client::shell::ClientEndpointFocusTarget>, + endpoints: &mut EndpointRegistry, + ) -> Result<(), String> { + self.focus = focus; + if let ActivationPhase::ActivatingTarget { + focus_request_id, + focus_acknowledged, + .. + } = &mut self.phase + { + *focus_acknowledged = self.focus.is_none() && focus_request_id.is_none(); + } else { + // The latest target is retained and will be sent immediately after source release. + return Ok(()); + } + self.send_latest_focus(endpoints) + } + + pub(crate) fn update_resize( + &mut self, + resize: crate::protocol::ClientMessage, + endpoints: &mut EndpointRegistry, + ) -> Result<(), String> { + resize_geometry(&resize) + .ok_or_else(|| "endpoint activation did not include a surface resize".to_owned())?; + self.resize = resize.clone(); + let restart_effects_fence = match &self.phase { + ActivationPhase::AwaitingPresentationEffects { + lease, completion, .. + } => Some((lease.clone(), (**completion).clone())), + _ => None, + }; + if let Some((lease, completion)) = restart_effects_fence { + if endpoints.send_to(&lease.endpoint_id, &resize) != EndpointSendOutcome::Sent { + return Err("pending endpoint resize could not be sent".into()); + } + return self.start_presentation_sync(endpoints, lease, completion); + } + match &mut self.phase { + ActivationPhase::ActivatingTarget { evidence, .. } + | ActivationPhase::RestoringSource { evidence, .. } + | ActivationPhase::SynchronizingPresentation { evidence, .. } => { + evidence.invalidate_surface() + } + _ => {} + } + let destination = match &self.phase { + ActivationPhase::ActivatingTarget { .. } => Some(&self.target.endpoint_id), + ActivationPhase::RestoringSource { .. } => Some(&self.source.endpoint_id), + ActivationPhase::SynchronizingPresentation { lease, .. } => Some(&lease.endpoint_id), + _ => None, + }; + if let Some(destination) = destination { + if endpoints.send_to(destination, &resize) != EndpointSendOutcome::Sent { + return Err("pending endpoint resize could not be sent".into()); + } + } + Ok(()) + } + + pub(crate) fn update_host_focus( + &mut self, + focused: bool, + endpoints: &mut EndpointRegistry, + ) -> Result<(), String> { + self.host_focused = focused; + let restart = self.presentation_restart(); + let destination = match &self.phase { + ActivationPhase::ActivatingTarget { .. } => Some(&self.target.endpoint_id), + ActivationPhase::RestoringSource { .. } => Some(&self.source.endpoint_id), + ActivationPhase::SynchronizingPresentation { lease, .. } + | ActivationPhase::AwaitingPresentationEffects { lease, .. } => { + Some(&lease.endpoint_id) + } + _ => None, + }; + if let Some(destination) = destination { + if endpoints.send_to( + destination, + &crate::protocol::ClientMessage::ClientShellFocus { focused }, + ) != EndpointSendOutcome::Sent + { + return Err("pending endpoint focus baseline could not be sent".into()); + } + } + if let Some((lease, completion)) = restart { + self.start_presentation_sync(endpoints, lease, completion)?; + } else { + self.invalidate_current_evidence(); + } + Ok(()) + } + + pub(crate) fn update_host_theme( + &mut self, + update: crate::protocol::ClientHostThemeUpdate, + endpoints: &mut EndpointRegistry, + ) -> Result<(), String> { + let restart = self.presentation_restart(); + let destination = match &self.phase { + ActivationPhase::ActivatingTarget { .. } => Some(&self.target.endpoint_id), + ActivationPhase::RestoringSource { .. } => Some(&self.source.endpoint_id), + ActivationPhase::SynchronizingPresentation { lease, .. } + | ActivationPhase::AwaitingPresentationEffects { lease, .. } => { + Some(&lease.endpoint_id) + } + _ => None, + }; + if let Some(destination) = destination { + let message = crate::protocol::ClientMessage::ClientShellHostTheme { update }; + if endpoints.send_to(destination, &message) != EndpointSendOutcome::Sent { + return Err("pending endpoint host theme could not be sent".into()); + } + } + if let Some((lease, completion)) = restart { + self.start_presentation_sync(endpoints, lease, completion)?; + } else { + self.invalidate_current_evidence(); + } + Ok(()) + } + + fn presentation_restart(&self) -> Option<(EndpointLease, ActivationCompletion)> { + match &self.phase { + ActivationPhase::SynchronizingPresentation { + lease, completion, .. + } + | ActivationPhase::AwaitingPresentationEffects { + lease, completion, .. + } => Some((lease.clone(), (**completion).clone())), + _ => None, + } + } + + fn invalidate_current_evidence(&mut self) { + match &mut self.phase { + ActivationPhase::ActivatingTarget { evidence, .. } + | ActivationPhase::RestoringSource { evidence, .. } => evidence.invalidate_surface(), + _ => {} + } + } + + /// Losing the source revokes its surface and removes the rollback destination; it must not + /// cancel a healthy target. Losing the target restores the source when it is still available. + pub(crate) fn endpoint_disconnected( + &mut self, + endpoints: &mut EndpointRegistry, + endpoint_id: &ClientEndpointId, + error: String, + ) -> ActivationRollback { + self.rollback_error = Some(error.clone()); + if self.target.endpoint_id == *endpoint_id && self.source.endpoint_id != *endpoint_id { + return match self.phase { + ActivationPhase::RestoringSource { .. } => ActivationRollback::Pending, + _ => match self.start_source_restore(endpoints, self.resize.clone()) { + Ok(()) => ActivationRollback::Pending, + Err(restore_error) => ActivationRollback::Unavailable(format!( + "{error}; source endpoint could not be restored safely: {restore_error}" + )), + }, + }; + } + if self.source.endpoint_id != *endpoint_id { + return ActivationRollback::Unavailable(error); + } + self.source_available = false; + if self.source.endpoint_id != self.target.endpoint_id { + match &self.phase { + ActivationPhase::ReleasingSource { .. } => { + return match self.start_target(endpoints, self.resize.clone()) { + Ok(()) => ActivationRollback::Pending, + Err(message) => ActivationRollback::Unavailable(message), + }; + } + ActivationPhase::ActivatingTarget { .. } => return ActivationRollback::Pending, + ActivationPhase::SynchronizingPresentation { lease, .. } + | ActivationPhase::AwaitingPresentationEffects { lease, .. } + if lease.endpoint_id == self.target.endpoint_id => + { + return ActivationRollback::Pending + } + _ => {} + } + } + match self.phase { + ActivationPhase::ReleasingSource { .. } => ActivationRollback::Unavailable(error), + ActivationPhase::ActivatingTarget { .. } => { + match self.start_target_release(endpoints) { + Ok(()) => ActivationRollback::Pending, + Err(release_error) => ActivationRollback::Unavailable(format!( + "{error}; target endpoint could not be released safely: {release_error}" + )), + } + } + ActivationPhase::ReleasingTargetForRollback { .. } => ActivationRollback::Pending, + ActivationPhase::RestoringSource { .. } => ActivationRollback::Unavailable(error), + ActivationPhase::SynchronizingPresentation { ref lease, .. } + | ActivationPhase::AwaitingPresentationEffects { ref lease, .. } => { + if lease.endpoint_id == self.target.endpoint_id { + match self.start_target_release(endpoints) { + Ok(()) => ActivationRollback::Pending, + Err(release_error) => ActivationRollback::Unavailable(format!( + "{error}; target endpoint could not be released safely: {release_error}" + )), + } + } else { + ActivationRollback::Unavailable(error) + } + } + } + } + + pub(crate) fn rollback( + &mut self, + endpoints: &mut EndpointRegistry, + error: String, + source_release_rejected: bool, + ) -> ActivationRollback { + self.rollback_error = Some(error.clone()); + if matches!(self.phase, ActivationPhase::ReleasingSource { .. }) && source_release_rejected + { + // Rejection proves source-off did not commit, but cached source metadata may have + // advanced while the frame was frozen. Restore through the same coherent on/sync + // path rather than immediately exposing a stale source projection. + return match self.start_source_restore(endpoints, self.resize.clone()) { + Ok(()) => ActivationRollback::Pending, + Err(restore_error) => ActivationRollback::Unavailable(format!( + "{error}; source endpoint could not resume: {restore_error}" + )), + }; + } + let result = match self.phase { + ActivationPhase::ReleasingSource { .. } => { + if self.source_available { + self.start_source_restore(endpoints, self.resize.clone()) + } else { + Err("the previous endpoint is no longer connected".into()) + } + } + ActivationPhase::ActivatingTarget { .. } => self.start_target_release(endpoints), + ActivationPhase::ReleasingTargetForRollback { .. } => { + // The target may have observed target-on or target-off. Closing this transport + // is the only safe local revocation when target-off is not acknowledged. + endpoints.fail( + &self.target.endpoint_id, + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "endpoint did not acknowledge surface revocation", + ), + ); + if !self.source_available { + return ActivationRollback::Unavailable(format!( + "{error}; the target connection was closed because no presentation owner could be proven" + )); + } + self.start_source_restore(endpoints, self.resize.clone()) + } + ActivationPhase::RestoringSource { .. } => { + return ActivationRollback::Unavailable(format!( + "{error}; source endpoint could not be restored" + )); + } + ActivationPhase::SynchronizingPresentation { ref lease, .. } + | ActivationPhase::AwaitingPresentationEffects { ref lease, .. } => { + if lease.endpoint_id == self.target.endpoint_id { + self.start_target_release(endpoints) + } else { + return ActivationRollback::Unavailable(format!( + "{error}; source endpoint presentation could not be synchronized" + )); + } + } + }; + match result { + Ok(()) => ActivationRollback::Pending, + Err(rollback_error) => ActivationRollback::Unavailable(format!( + "{error}; source endpoint could not be restored safely: {rollback_error}" + )), + } + } + + pub(crate) fn complete( + &mut self, + shell: &mut crate::client::shell::ClientShellState, + endpoints: &mut EndpointRegistry, + ) -> Result<ActivationCompletion, String> { + if let ActivationPhase::SynchronizingPresentation { + lease, + acknowledged_revision, + evidence, + completion, + .. + } = &self.phase + { + let lease = lease.clone(); + let completion = (**completion).clone(); + let surface = coherent_completion_surface( + shell, + &lease, + evidence, + *acknowledged_revision, + self.geometry(), + )?; + if endpoints.active_id() != &lease.endpoint_id + || !shell.endpoint_projection_available(&lease.endpoint_id) + || !shell.activate_endpoint_projection(&lease.endpoint_id) + { + return Err( + "endpoint became unavailable during presentation synchronization".into(), + ); + } + shell.set_pane_surface(surface); + self.start_presentation_effects_fence(endpoints, lease, completion)?; + return Ok(ActivationCompletion::AwaitingPresentationEffects); + } + if let ActivationPhase::AwaitingPresentationEffects { + ready: true, + completion, + .. + } = &self.phase + { + return Ok((**completion).clone()); + } + + let (lease, evidence, acknowledgement_revision, completion) = match &self.phase { + ActivationPhase::ActivatingTarget { + evidence, + acknowledged_revision, + .. + } => { + let completion = ActivationCompletion::Activated; + ( + self.target.clone(), + evidence, + *acknowledged_revision, + completion, + ) + } + ActivationPhase::RestoringSource { + evidence, + acknowledged_revision, + .. + } => { + let completion = ActivationCompletion::RestoredSource { + error: self + .rollback_error + .clone() + .unwrap_or_else(|| "endpoint handoff was rolled back".into()), + successor: self.successor.clone(), + }; + ( + self.source.clone(), + evidence, + *acknowledged_revision, + completion, + ) + } + _ => return Err("endpoint activation completed in an invalid phase".into()), + }; + let surface = coherent_completion_surface( + shell, + &lease, + evidence, + acknowledgement_revision, + self.geometry(), + )?; + endpoints.set_surface_active(&lease.endpoint_id, true); + shell.set_endpoint_status(&lease.endpoint_id, ClientEndpointStatus::Online); + if !shell.endpoint_projection_available(&lease.endpoint_id) + || !endpoints.set_active(&lease.endpoint_id) + { + return Err("endpoint became unavailable during activation".into()); + } + let activated = shell.activate_endpoint_projection(&lease.endpoint_id); + debug_assert!(activated, "preflighted endpoint projection must activate"); + debug_assert!(shell.endpoint_is_active(endpoints.active_id())); + shell.set_pane_surface(surface); + self.start_presentation_sync(endpoints, lease.clone(), completion)?; + Ok(ActivationCompletion::AwaitingPresentationSync { + previous: self.source.endpoint_id.clone(), + endpoint: lease.endpoint_id, + }) + } + + fn start_target( + &mut self, + endpoints: &mut EndpointRegistry, + resize: crate::protocol::ClientMessage, + ) -> Result<(), String> { + let request_id = format!("client-shell-surface:{}:on", self.epoch); + self.deadline = Instant::now() + ACTIVATION_TIMEOUT; + // A transport may fail after writing any baseline or surface message. Enter the target + // phase first so every uncertain target write is reversed through target-off before + // source restoration is considered. + self.phase = ActivationPhase::ActivatingTarget { + request_id: request_id.clone(), + acknowledged_revision: None, + focus_request_id: None, + focus_request_target: None, + focus_acknowledged: self.focus.is_none(), + evidence: ActivationEvidence::default(), + }; + send_surface_activation( + endpoints, + &self.target, + request_id, + &resize, + self.host_focused, + )?; + + // From this point the target may have processed surface.set(true). Optional navigation + // is serialized through one coalescing focus lane. + self.send_latest_focus(endpoints) + } + + fn start_presentation_sync( + &mut self, + endpoints: &mut EndpointRegistry, + lease: EndpointLease, + completion: ActivationCompletion, + ) -> Result<(), String> { + let request_id = format!("client-shell-surface:{}:presentation-sync", self.epoch); + let request = surface_interest_request(&lease.boot_id, request_id.clone(), true) + .map_err(|error| error.to_string())?; + self.phase = ActivationPhase::SynchronizingPresentation { + lease: lease.clone(), + request_id, + acknowledged_revision: None, + evidence: ActivationEvidence::default(), + completion: Box::new(completion), + }; + self.deadline = Instant::now() + ACTIVATION_TIMEOUT; + if endpoints.send_to(&lease.endpoint_id, &request) != EndpointSendOutcome::Sent { + return Err("endpoint presentation synchronization could not be sent".into()); + } + Ok(()) + } + + fn start_presentation_effects_fence( + &mut self, + endpoints: &mut EndpointRegistry, + lease: EndpointLease, + completion: ActivationCompletion, + ) -> Result<(), String> { + let token = format!("{}:{}:{}", self.epoch, lease.generation, lease.boot_id); + self.phase = ActivationPhase::AwaitingPresentationEffects { + lease: lease.clone(), + token: token.clone(), + ready: false, + completion: Box::new(completion), + }; + self.deadline = Instant::now() + ACTIVATION_TIMEOUT; + let message = crate::protocol::ClientMessage::EndpointControl { + kind: crate::protocol::endpoint::PRESENTATION_EFFECTS_SYNC_KIND.into(), + data: token, + }; + if endpoints.send_to(&lease.endpoint_id, &message) != EndpointSendOutcome::Sent { + return Err("endpoint presentation effects fence could not be sent".into()); + } + Ok(()) + } + + fn start_target_release(&mut self, endpoints: &mut EndpointRegistry) -> Result<(), String> { + let request_id = format!("client-shell-surface:{}:rollback-target-off", self.epoch); + let request = surface_interest_request(&self.target.boot_id, request_id.clone(), false) + .map_err(|error| error.to_string())?; + // Set the rollback phase before the potentially observed target-off write. + self.phase = ActivationPhase::ReleasingTargetForRollback { request_id }; + self.deadline = Instant::now() + ACTIVATION_TIMEOUT; + if endpoints.send_to(&self.target.endpoint_id, &request) != EndpointSendOutcome::Sent { + return Err("target endpoint release could not be sent".into()); + } + Ok(()) + } + + fn start_source_restore( + &mut self, + endpoints: &mut EndpointRegistry, + resize: crate::protocol::ClientMessage, + ) -> Result<(), String> { + let request_id = format!("client-shell-surface:{}:rollback-source-on", self.epoch); + // Source baseline writes can also be observed before their send reports an error. + self.phase = ActivationPhase::RestoringSource { + request_id: request_id.clone(), + acknowledged_revision: None, + evidence: ActivationEvidence::default(), + }; + self.deadline = Instant::now() + ACTIVATION_TIMEOUT; + send_surface_activation( + endpoints, + &self.source, + request_id, + &resize, + self.host_focused, + ) + } + + fn send_latest_focus(&mut self, endpoints: &mut EndpointRegistry) -> Result<(), String> { + let desired = self.focus.clone(); + let Some(desired) = desired else { + if let ActivationPhase::ActivatingTarget { + focus_request_id, + focus_acknowledged, + .. + } = &mut self.phase + { + if focus_request_id.is_none() { + *focus_acknowledged = true; + } + } + return Ok(()); + }; + if !matches!( + &self.phase, + ActivationPhase::ActivatingTarget { + focus_request_id: None, + .. + } + ) { + return Ok(()); + } + let request_id = self + .next_focus_request_id() + .expect("desired focus creates a request id"); + if let ActivationPhase::ActivatingTarget { + focus_request_id, + focus_request_target, + focus_acknowledged, + .. + } = &mut self.phase + { + *focus_acknowledged = false; + *focus_request_id = Some(request_id.clone()); + *focus_request_target = Some(desired.clone()); + } + let request = focus_request(&self.target.boot_id, request_id, &desired) + .map_err(|error| error.to_string())?; + if endpoints.send_to(&self.target.endpoint_id, &request) != EndpointSendOutcome::Sent { + return Err("endpoint focus could not be sent".into()); + } + Ok(()) + } + + fn next_focus_request_id(&mut self) -> Option<String> { + self.focus.as_ref()?; + self.next_focus_serial = self.next_focus_serial.saturating_add(1); + Some(format!( + "client-shell-focus:{}:{}", + self.epoch, self.next_focus_serial + )) + } + + fn progress(&self) -> SurfaceActivationProgress { + match &self.phase { + ActivationPhase::ActivatingTarget { + acknowledged_revision, + focus_acknowledged, + evidence, + .. + } if acknowledged_revision.is_some_and(|revision| { + *focus_acknowledged + && evidence + .coherent_surface(revision, self.geometry()) + .is_some_and(|surface| self.target_matches(surface)) + }) => + { + SurfaceActivationProgress::Ready + } + ActivationPhase::RestoringSource { + acknowledged_revision, + evidence, + .. + } + | ActivationPhase::SynchronizingPresentation { + acknowledged_revision, + evidence, + .. + } if acknowledged_revision.is_some_and(|revision| { + evidence + .coherent_surface(revision, self.geometry()) + .is_some() + }) => + { + SurfaceActivationProgress::Ready + } + _ => SurfaceActivationProgress::Pending, + } + } + + fn target_matches(&self, surface: &crate::protocol::PaneSurfaceFrame) -> bool { + let evidence = match &self.phase { + ActivationPhase::ActivatingTarget { evidence, .. } => evidence, + _ => return false, + }; + match &self.focus { + Some(crate::client::shell::ClientEndpointFocusTarget::Pane(pane_id)) => { + evidence.focused_pane_id.as_deref() == Some(pane_id) + && surface + .panes + .iter() + .any(|pane| pane.focused && &pane.pane_id == pane_id) + } + Some(crate::client::shell::ClientEndpointFocusTarget::Tab(tab_id)) => { + evidence.focused_tab_id.as_deref() == Some(tab_id) + } + Some(crate::client::shell::ClientEndpointFocusTarget::Workspace(workspace_id)) => { + evidence.focused_workspace_id.as_deref() == Some(workspace_id) + } + None => true, + } + } +} + +#[cfg(test)] +#[path = "activation_tests.rs"] +mod tests; diff --git a/src/client/endpoint/activation/model.rs b/src/client/endpoint/activation/model.rs new file mode 100644 index 00000000..c94029e9 --- /dev/null +++ b/src/client/endpoint/activation/model.rs @@ -0,0 +1,174 @@ +use std::time::Instant; + +use super::super::ClientEndpointId; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct EndpointLease { + pub(super) endpoint_id: ClientEndpointId, + pub(super) generation: u64, + pub(super) boot_id: String, + /// The endpoint cache may already be newer than the first activation event. Never let an + /// activation prove coherence with a revision that the monotonic cache has discarded. + pub(super) minimum_revision: u64, +} + +#[derive(Clone, Debug, Default)] +pub(super) struct ActivationEvidence { + pub(super) snapshot_revision: Option<u64>, + pub(super) focused_workspace_id: Option<String>, + pub(super) focused_tab_id: Option<String>, + pub(super) focused_pane_id: Option<String>, + pub(super) surface: Option<crate::protocol::PaneSurfaceFrame>, +} + +impl ActivationEvidence { + pub(super) fn record_snapshot(&mut self, snapshot: &crate::protocol::ClientShellSnapshot) { + if self + .snapshot_revision + .is_none_or(|current| snapshot.revision >= current) + { + self.snapshot_revision = Some(snapshot.revision); + self.focused_workspace_id = snapshot.focused_workspace_id.clone(); + self.focused_tab_id = snapshot.focused_tab_id.clone(); + self.focused_pane_id = snapshot.focused_pane_id.clone(); + } + } + + pub(super) fn record_surface(&mut self, surface: crate::protocol::PaneSurfaceFrame) { + let replace = self.surface.as_ref().is_none_or(|current| { + surface.projection_revision > current.projection_revision + || (surface.projection_revision == current.projection_revision + && surface.surface_revision >= current.surface_revision) + }); + if replace { + self.surface = Some(surface); + } + } + + pub(super) fn invalidate_surface(&mut self) { + self.surface = None; + } + + pub(super) fn coherent_surface( + &self, + minimum_revision: u64, + geometry: crate::protocol::ClientSurfaceSize, + ) -> Option<&crate::protocol::PaneSurfaceFrame> { + self.surface.as_ref().filter(|surface| { + self.snapshot_revision == Some(surface.projection_revision) + && surface.projection_revision >= minimum_revision + && super::surface_matches_geometry(surface, geometry) + }) + } +} + +#[derive(Clone, Debug)] +pub(super) enum ActivationPhase { + ReleasingSource { + request_id: String, + }, + ActivatingTarget { + request_id: String, + acknowledged_revision: Option<u64>, + /// At most one focus request may be in flight. Retargets only replace `focus` until + /// this response arrives, at which point the latest target is sent. + focus_request_id: Option<String>, + focus_request_target: Option<crate::client::shell::ClientEndpointFocusTarget>, + focus_acknowledged: bool, + evidence: ActivationEvidence, + }, + ReleasingTargetForRollback { + request_id: String, + }, + RestoringSource { + request_id: String, + acknowledged_revision: Option<u64>, + evidence: ActivationEvidence, + }, + SynchronizingPresentation { + lease: EndpointLease, + request_id: String, + acknowledged_revision: Option<u64>, + evidence: ActivationEvidence, + completion: Box<ActivationCompletion>, + }, + AwaitingPresentationEffects { + lease: EndpointLease, + token: String, + ready: bool, + completion: Box<ActivationCompletion>, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum SurfaceActivationProgress { + Pending, + Ready, + Rejected { + message: String, + source_release_rejected: bool, + }, + Stale, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum ActivationRollback { + /// A correlated rollback request is in flight; retain the frozen frame. + Pending, + /// Neither endpoint can be made safe to present. Keep pane input frozen while presenting + /// client chrome and this error. + Unavailable(String), +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum ActivationCompletion { + AwaitingPresentationSync { + previous: ClientEndpointId, + endpoint: ClientEndpointId, + }, + AwaitingPresentationEffects, + Activated, + RestoredSource { + error: String, + /// A newer endpoint-qualified selection arrived while this handoff was frozen. It is + /// started only after the original source has been coherently restored. + successor: Option<EndpointActivationIntent>, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct EndpointActivationIntent { + pub(crate) endpoint_id: ClientEndpointId, + pub(crate) target: Option<crate::client::shell::ClientEndpointFocusTarget>, +} + +/// Begin failures are separated by whether the transport may already have observed a lifecycle +/// write. A partial transaction must be kept and rolled back under a frozen presentation. +#[derive(Debug)] +pub(crate) enum ActivationBeginError { + Preflight(String), + Partial { + activation: Box<PendingEndpointActivation>, + error: String, + }, +} + +/// The only owner of an endpoint handoff. The registry's active endpoint remains the committed +/// endpoint until the target commits, while this object owns the uncommitted lifecycle lane. +#[derive(Debug)] +pub(crate) struct PendingEndpointActivation { + pub(super) source: EndpointLease, + pub(super) source_available: bool, + pub(super) target: EndpointLease, + pub(super) focus: Option<crate::client::shell::ClientEndpointFocusTarget>, + pub(super) host_focused: bool, + pub(super) resize: crate::protocol::ClientMessage, + pub(super) phase: ActivationPhase, + pub(super) deadline: Instant, + pub(super) epoch: u64, + pub(super) next_focus_serial: u64, + pub(super) rollback_error: Option<String>, + /// A different endpoint was selected while this source-off-first transaction was in flight. + /// Keep only the latest intent until source restoration commits. + pub(super) successor: Option<EndpointActivationIntent>, +} diff --git a/src/client/endpoint/activation/protocol.rs b/src/client/endpoint/activation/protocol.rs new file mode 100644 index 00000000..569d53f3 --- /dev/null +++ b/src/client/endpoint/activation/protocol.rs @@ -0,0 +1,219 @@ +use super::super::{ClientEndpointId, EndpointRegistry, EndpointSendOutcome}; +use super::model::{ActivationEvidence, EndpointLease}; + +pub(super) fn focus_result_matches( + focus: Option<&crate::client::shell::ClientEndpointFocusTarget>, + result: &crate::api::schema::ResponseResult, +) -> bool { + match (focus, result) { + ( + Some(crate::client::shell::ClientEndpointFocusTarget::Pane(expected)), + crate::api::schema::ResponseResult::PaneInfo { pane }, + ) => pane.focused && &pane.pane_id == expected, + ( + Some(crate::client::shell::ClientEndpointFocusTarget::Workspace(expected)), + crate::api::schema::ResponseResult::WorkspaceInfo { workspace }, + ) => workspace.focused && &workspace.workspace_id == expected, + ( + Some(crate::client::shell::ClientEndpointFocusTarget::Tab(expected)), + crate::api::schema::ResponseResult::TabInfo { tab }, + ) => tab.focused && &tab.tab_id == expected, + _ => false, + } +} + +pub(super) fn endpoint_lease( + shell: &crate::client::shell::ClientShellState, + endpoints: &EndpointRegistry, + endpoint_id: &ClientEndpointId, +) -> Result<EndpointLease, String> { + let connection = endpoints + .connection(endpoint_id) + .ok_or_else(|| "endpoint connection is unavailable".to_owned())?; + let (boot_id, minimum_revision) = shell + .endpoint_snapshot_identity(endpoint_id, connection.generation) + .ok_or_else(|| "endpoint metadata is not ready for this connection".to_owned())?; + Ok(EndpointLease { + endpoint_id: endpoint_id.clone(), + generation: connection.generation, + boot_id: boot_id.to_owned(), + minimum_revision, + }) +} + +pub(super) fn disconnected_endpoint_lease( + shell: &crate::client::shell::ClientShellState, + endpoint_id: &ClientEndpointId, +) -> EndpointLease { + EndpointLease { + endpoint_id: endpoint_id.clone(), + generation: 0, + boot_id: shell + .endpoint_boot_id(endpoint_id) + .unwrap_or_default() + .to_owned(), + minimum_revision: 0, + } +} + +pub(super) fn endpoint_matches( + lease: &EndpointLease, + endpoint_id: &ClientEndpointId, + generation: u64, + boot_id: &str, +) -> bool { + lease.endpoint_id == *endpoint_id && lease.generation == generation && lease.boot_id == boot_id +} + +pub(super) fn coherent_completion_surface( + shell: &crate::client::shell::ClientShellState, + lease: &EndpointLease, + evidence: &ActivationEvidence, + acknowledgement_revision: Option<u64>, + geometry: crate::protocol::ClientSurfaceSize, +) -> Result<crate::protocol::PaneSurfaceFrame, String> { + let acknowledgement_revision = acknowledgement_revision.ok_or_else(|| { + "endpoint activation completed without a surface acknowledgement".to_owned() + })?; + let surface = evidence + .surface + .clone() + .ok_or_else(|| "endpoint activation completed without a surface".to_owned())?; + if surface.projection_revision < acknowledgement_revision + || !surface_matches_geometry(&surface, geometry) + { + return Err("endpoint activation lost its acknowledged surface evidence".into()); + } + if !shell.endpoint_snapshot_matches( + &lease.endpoint_id, + lease.generation, + &lease.boot_id, + surface.projection_revision, + ) { + return Err("endpoint activation lost its coherent snapshot/surface pair".into()); + } + Ok(surface) +} + +pub(super) fn resize_geometry( + message: &crate::protocol::ClientMessage, +) -> Option<crate::protocol::ClientSurfaceSize> { + match message { + crate::protocol::ClientMessage::ClientShellResize { surface_size, .. } => { + Some(*surface_size) + } + _ => None, + } +} + +pub(super) fn surface_matches_geometry( + surface: &crate::protocol::PaneSurfaceFrame, + geometry: crate::protocol::ClientSurfaceSize, +) -> bool { + surface.frame.width == geometry.cols && surface.frame.height == geometry.rows +} + +pub(super) fn send_surface_activation( + endpoints: &mut EndpointRegistry, + target: &EndpointLease, + request_id: String, + resize: &crate::protocol::ClientMessage, + focused: bool, +) -> Result<(), String> { + if endpoints.send_to(&target.endpoint_id, resize) != EndpointSendOutcome::Sent { + return Err("endpoint resize could not be sent".into()); + } + let request = surface_interest_request(&target.boot_id, request_id, true) + .map_err(|error| error.to_string())?; + if endpoints.send_to(&target.endpoint_id, &request) != EndpointSendOutcome::Sent { + return Err("endpoint activation could not be sent".into()); + } + // Inactive endpoints reject focus events. Activate first, then establish the host baseline + // on the same ordered transport before navigation or presentation can commit. + if endpoints.send_to( + &target.endpoint_id, + &crate::protocol::ClientMessage::ClientShellFocus { focused }, + ) != EndpointSendOutcome::Sent + { + return Err("endpoint focus baseline could not be sent".into()); + } + Ok(()) +} + +pub(super) fn decode_endpoint_response( + request_id: &str, + data: &[u8], +) -> Result<crate::api::schema::ResponseResult, crate::client::shell::ClientShellEndpointError> { + crate::client::endpoint_commands::parse_response(request_id, data) +} + +pub(super) fn surface_set_revision( + result: &crate::api::schema::ResponseResult, + expected_active: bool, +) -> Result<u64, String> { + match result { + crate::api::schema::ResponseResult::ClientShellSurfaceSet { + active, + projection_revision, + } if *active == expected_active => Ok(*projection_revision), + _ => Err("surface activation returned an invalid acknowledgement".into()), + } +} + +pub(super) fn focus_request( + boot_id: &str, + request_id: String, + focus: &crate::client::shell::ClientEndpointFocusTarget, +) -> std::io::Result<crate::protocol::ClientMessage> { + let method = match focus { + crate::client::shell::ClientEndpointFocusTarget::Workspace(workspace_id) => { + crate::api::schema::Method::WorkspaceFocus(crate::api::schema::WorkspaceTarget { + workspace_id: workspace_id.clone(), + }) + } + crate::client::shell::ClientEndpointFocusTarget::Pane(pane_id) => { + crate::api::schema::Method::PaneFocus(crate::api::schema::PaneTarget { + pane_id: pane_id.clone(), + }) + } + crate::client::shell::ClientEndpointFocusTarget::Tab(tab_id) => { + crate::api::schema::Method::TabFocus(crate::api::schema::TabTarget { + tab_id: tab_id.clone(), + }) + } + }; + endpoint_request( + boot_id, + crate::api::schema::Request { + id: request_id, + method, + }, + ) +} + +pub(super) fn surface_interest_request( + boot_id: &str, + request_id: String, + active: bool, +) -> std::io::Result<crate::protocol::ClientMessage> { + endpoint_request( + boot_id, + crate::api::schema::Request { + id: request_id, + method: crate::api::schema::Method::ClientShellSurfaceSet( + crate::api::schema::ClientShellSurfaceSetParams { active }, + ), + }, + ) +} + +fn endpoint_request( + boot_id: &str, + request: crate::api::schema::Request, +) -> std::io::Result<crate::protocol::ClientMessage> { + Ok(crate::protocol::ClientMessage::ClientShellEndpointRequest { + boot_id: boot_id.to_owned(), + request: serde_json::to_string(&request) + .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?, + }) +} diff --git a/src/client/endpoint/activation_tests.rs b/src/client/endpoint/activation_tests.rs new file mode 100644 index 00000000..b66cb98b --- /dev/null +++ b/src/client/endpoint/activation_tests.rs @@ -0,0 +1,1305 @@ +use super::*; + +fn endpoint() -> ClientEndpointId { + ClientEndpointId::Ssh( + super::super::ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap(), + ) +} + +fn lease(id: ClientEndpointId, generation: u64, boot: &str) -> EndpointLease { + EndpointLease { + endpoint_id: id, + generation, + boot_id: boot.into(), + minimum_revision: 0, + } +} + +#[derive(Clone)] +struct FakeTransport { + sent: std::sync::Arc<std::sync::Mutex<Vec<crate::protocol::ClientMessage>>>, + fail_after_write: bool, +} + +impl super::super::EndpointTransport for FakeTransport { + fn send(&mut self, message: &crate::protocol::ClientMessage) -> std::io::Result<()> { + self.sent.lock().unwrap().push(message.clone()); + if self.fail_after_write { + Err(std::io::Error::other("simulated observed write failure")) + } else { + Ok(()) + } + } +} + +fn negotiation() -> super::super::EndpointNegotiation { + super::super::EndpointNegotiation::new( + vec!["client_shell.surface.set".into()], + vec![ + crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY.into(), + crate::protocol::endpoint::PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(), + ], + ) +} + +fn test_snapshot(boot_id: &str, revision: u64) -> crate::protocol::ClientShellSnapshot { + crate::protocol::ClientShellSnapshot { + boot_id: boot_id.into(), + revision, + config_diagnostic: None, + product_announcement: None, + update_available: None, + update_install_command: String::new(), + server_keybindings_toml: None, + latest_release_notes_available: false, + integration_updates_available: false, + worktree_directory: String::new(), + release_notes: None, + focused_workspace_id: None, + focused_tab_id: None, + focused_pane_id: None, + tab_bar_right: Vec::new(), + tab_bar_right_separator: String::new(), + agent_view_label: None, + agent_order: Vec::new(), + workspaces: Vec::new(), + tabs: Vec::new(), + panes: Vec::new(), + agents: Vec::new(), + commands: Vec::new(), + } +} + +type SentMessages = std::sync::Arc<std::sync::Mutex<Vec<crate::protocol::ClientMessage>>>; +type TestFixture = ( + crate::client::ClientShellState, + EndpointRegistry, + SentMessages, + SentMessages, +); + +fn shell_and_registry() -> TestFixture { + shell_and_registry_with_source_failure(false) +} + +fn shell_and_registry_with_source_failure(source_fail_after_write: bool) -> TestFixture { + let mut shell = crate::client::ClientShellState::new( + crate::client::ClientShellConfig::from_config(&crate::config::Config::default()), + ); + let profile = super::super::SavedSshEndpoint { + id: super::super::ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap(), + label: "Remote".into(), + target: "dev@example.com".into(), + session: "main".into(), + enabled: true, + }; + let target = ClientEndpointId::Ssh(profile.id.clone()); + shell.set_endpoint_catalog(&[profile]); + shell.set_snapshot(Box::new(test_snapshot("local-boot", 1))); + shell.set_endpoint_status(&target, ClientEndpointStatus::Online); + shell.set_endpoint_snapshot(&target, Box::new(test_snapshot("remote-boot", 1))); + + let local_sent = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let remote_sent = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let mut endpoints = EndpointRegistry::new( + FakeTransport { + sent: local_sent.clone(), + fail_after_write: source_fail_after_write, + }, + 1, + negotiation(), + ); + endpoints.insert( + target, + FakeTransport { + sent: remote_sent.clone(), + fail_after_write: false, + }, + 7, + negotiation(), + false, + ); + (shell, endpoints, local_sent, remote_sent) +} + +fn surface_success(id: &str, active: bool, projection_revision: u64) -> Vec<u8> { + serde_json::to_vec(&crate::api::schema::SuccessResponse { + id: id.into(), + result: crate::api::schema::ResponseResult::ClientShellSurfaceSet { + active, + projection_revision, + }, + }) + .unwrap() +} + +fn workspace_focus_success(id: &str, workspace_id: &str) -> Vec<u8> { + serde_json::to_vec(&crate::api::schema::SuccessResponse { + id: id.into(), + result: crate::api::schema::ResponseResult::WorkspaceInfo { + workspace: crate::api::schema::WorkspaceInfo { + workspace_id: workspace_id.into(), + number: 1, + label: workspace_id.into(), + focused: true, + pane_count: 1, + tab_count: 1, + active_tab_id: "tab".into(), + agent_status: crate::api::schema::AgentStatus::Unknown, + tokens: Default::default(), + worktree: None, + }, + }, + }) + .unwrap() +} + +fn failure(id: &str, message: &str) -> Vec<u8> { + serde_json::to_vec(&crate::api::schema::ErrorResponse { + id: id.into(), + error: crate::api::schema::ErrorBody { + code: "surface_rejected".into(), + message: message.into(), + }, + }) + .unwrap() +} + +fn surface_set_active(message: &crate::protocol::ClientMessage) -> Option<bool> { + let crate::protocol::ClientMessage::ClientShellEndpointRequest { request, .. } = message else { + return None; + }; + let request: crate::api::schema::Request = serde_json::from_str(request).ok()?; + match request.method { + crate::api::schema::Method::ClientShellSurfaceSet(params) => Some(params.active), + _ => None, + } +} + +fn resize() -> crate::protocol::ClientMessage { + crate::protocol::ClientMessage::ClientShellResize { + cell_width_px: 8, + cell_height_px: 16, + surface_size: crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 }, + pixel_mouse: false, + } +} + +fn surface(boot_id: &str, revision: u64, pane: &str) -> crate::protocol::PaneSurfaceFrame { + crate::protocol::PaneSurfaceFrame { + boot_id: boot_id.into(), + projection_revision: revision, + surface_revision: revision, + frame: crate::protocol::FrameData { + cells: Vec::new(), + width: 80, + height: 24, + cursor: None, + hyperlinks: Vec::new(), + graphics: Vec::new(), + }, + panes: vec![crate::protocol::PaneSurfacePane { + pane_id: pane.into(), + content_revision: revision, + rect: crate::protocol::SurfaceRect { + x: 0, + y: 0, + width: 80, + height: 24, + }, + inner_rect: crate::protocol::SurfaceRect { + x: 0, + y: 0, + width: 80, + height: 24, + }, + scrollbar_rect: None, + scroll: None, + focused: true, + mouse_reporting: false, + sgr_pixel_mouse: false, + alternate_screen_active: false, + pixel_width: 0, + pixel_height: 0, + }], + splits: Vec::new(), + popup: None, + graphics: Default::default(), + } +} + +fn machine() -> PendingEndpointActivation { + PendingEndpointActivation { + source: lease(ClientEndpointId::Local, 1, "local-boot"), + source_available: true, + target: lease(endpoint(), 7, "remote-boot"), + focus: None, + host_focused: true, + resize: resize(), + phase: ActivationPhase::ActivatingTarget { + request_id: "client-shell-surface:3:on".into(), + acknowledged_revision: Some(1), + focus_request_id: None, + focus_request_target: None, + focus_acknowledged: true, + evidence: ActivationEvidence::default(), + }, + deadline: Instant::now() + ACTIVATION_TIMEOUT, + epoch: 3, + next_focus_serial: 0, + rollback_error: None, + successor: None, + } +} + +#[test] +fn source_off_request_is_distinct_and_precedes_target_on_phase() { + let activation = PendingEndpointActivation { + source: lease(ClientEndpointId::Local, 1, "local-boot"), + source_available: true, + target: lease(endpoint(), 7, "remote-boot"), + focus: None, + host_focused: true, + resize: resize(), + phase: ActivationPhase::ReleasingSource { + request_id: "client-shell-surface:9:off".into(), + }, + deadline: Instant::now() + ACTIVATION_TIMEOUT, + epoch: 9, + next_focus_serial: 0, + rollback_error: None, + successor: None, + }; + assert!(activation.accepts_response( + &ClientEndpointId::Local, + 1, + "local-boot", + "client-shell-surface:9:off" + )); + assert!(!activation.accepts_response( + &endpoint(), + 7, + "remote-boot", + "client-shell-surface:9:on" + )); +} + +#[test] +fn active_source_requires_metadata_from_its_current_connection_generation() { + let (mut shell, mut endpoints, local_sent, remote_sent) = shell_and_registry(); + shell.set_endpoint_snapshot_for_generation( + &ClientEndpointId::Local, + 99, + Box::new(test_snapshot("stale-local-boot", 1)), + ); + + let result = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 26, + Instant::now(), + ); + + assert!( + matches!(result, Err(ActivationBeginError::Preflight(message)) if message.contains("this connection")) + ); + assert!(local_sent.lock().unwrap().is_empty()); + assert!(remote_sent.lock().unwrap().is_empty()); + assert!(endpoints.active_surface_available()); +} + +#[test] +fn observed_begin_write_failure_returns_recoverable_partial_activation() { + let (shell, mut endpoints, local_sent, remote_sent) = + shell_and_registry_with_source_failure(true); + let result = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 10, + Instant::now(), + ); + + let ActivationBeginError::Partial { + mut activation, + error, + } = (match result { + Err(error) => error, + Ok(_) => panic!("an observed source write must return partial lifecycle state"), + }) + else { + panic!("an observed source write must return partial lifecycle state"); + }; + assert!(error.contains("focus revoke")); + assert_eq!( + local_sent.lock().unwrap().first(), + Some(&crate::protocol::ClientMessage::ClientShellFocus { focused: false }) + ); + assert!(remote_sent.lock().unwrap().is_empty()); + assert!(matches!( + activation.rollback(&mut endpoints, error, false), + ActivationRollback::Unavailable(_) + )); +} + +#[test] +fn source_release_is_sent_and_acknowledged_before_target_activation() { + let (shell, mut endpoints, local_sent, remote_sent) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + None, + resize(), + 11, + Instant::now(), + ) + .unwrap(); + let local = local_sent.lock().unwrap(); + assert_eq!( + local.first(), + Some(&crate::protocol::ClientMessage::ClientShellFocus { focused: false }), + "source focus is revoked before source-off" + ); + assert_eq!( + local + .as_slice() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![false], + "the source is released first" + ); + drop(local); + assert!(remote_sent.lock().unwrap().is_empty()); + assert!( + !endpoints.active_surface_available(), + "pane input is blocked while frozen" + ); + + assert_eq!( + activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:11:off", + &surface_success("client-shell-surface:11:off", false, 1), + &mut endpoints, + ), + SurfaceActivationProgress::Pending + ); + let remote = remote_sent.lock().unwrap(); + assert!(matches!( + remote[0], + crate::protocol::ClientMessage::ClientShellResize { .. } + )); + assert_eq!( + remote.get(2), + Some(&crate::protocol::ClientMessage::ClientShellFocus { focused: true }) + ); + assert_eq!(remote.get(1).and_then(surface_set_active), Some(true)); +} + +#[test] +fn activation_requires_an_exact_snapshot_surface_revision_pair() { + let mut activation = machine(); + let target = endpoint(); + let snapshot = crate::protocol::ClientShellSnapshot { + boot_id: "remote-boot".into(), + revision: 2, + config_diagnostic: None, + product_announcement: None, + update_available: None, + update_install_command: String::new(), + server_keybindings_toml: None, + latest_release_notes_available: false, + integration_updates_available: false, + worktree_directory: String::new(), + release_notes: None, + focused_workspace_id: None, + focused_tab_id: None, + focused_pane_id: None, + tab_bar_right: Vec::new(), + tab_bar_right_separator: String::new(), + agent_view_label: None, + agent_order: Vec::new(), + workspaces: Vec::new(), + tabs: Vec::new(), + panes: Vec::new(), + agents: Vec::new(), + commands: Vec::new(), + }; + assert_eq!( + activation.receive_snapshot(&target, 7, &snapshot), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface(&target, 7, surface("remote-boot", 1, "pane")), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface(&target, 7, surface("remote-boot", 2, "pane")), + SurfaceActivationProgress::Ready + ); +} + +#[test] +fn typed_target_ack_sets_a_floor_for_same_boot_activation_evidence() { + let (shell, mut endpoints, _local_sent, _remote_sent) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + None, + resize(), + 16, + Instant::now(), + ) + .unwrap(); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:16:off", + &surface_success("client-shell-surface:16:off", false, 1), + &mut endpoints, + ); + assert_eq!( + activation.receive_response( + &target, + 7, + "client-shell-surface:16:on", + &surface_success("client-shell-surface:16:on", true, 4), + &mut endpoints, + ), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_snapshot(&target, 7, &test_snapshot("remote-boot", 3)), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface(&target, 7, surface("remote-boot", 3, "pane")), + SurfaceActivationProgress::Pending, + "a delayed same-boot surface below the acknowledgement floor is not evidence" + ); + assert_eq!( + activation.receive_snapshot(&target, 7, &test_snapshot("remote-boot", 4)), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface(&target, 7, surface("remote-boot", 4, "pane")), + SurfaceActivationProgress::Ready + ); +} + +#[test] +fn stale_generation_and_boot_are_not_activation_evidence() { + let mut activation = machine(); + assert_eq!( + activation.receive_surface(&endpoint(), 6, surface("remote-boot", 1, "pane")), + SurfaceActivationProgress::Stale + ); + assert_eq!( + activation.receive_surface(&endpoint(), 7, surface("old-boot", 1, "pane")), + SurfaceActivationProgress::Stale + ); +} + +#[test] +fn stale_response_boot_is_not_consumed() { + let mut activation = machine(); + let (_shell, mut endpoints, _local_sent, _remote_sent) = shell_and_registry(); + assert_eq!( + activation.receive_response_for_boot( + &endpoint(), + 7, + "old-boot", + "client-shell-surface:3:on", + &surface_success("client-shell-surface:3:on", true, 2), + &mut endpoints, + ), + SurfaceActivationProgress::Stale + ); +} + +#[test] +fn same_target_retarget_is_latest_wins() { + let (shell, mut endpoints, _local_sent, remote_sent) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + Some(crate::client::shell::ClientEndpointFocusTarget::Workspace( + "old".into(), + )), + resize(), + 12, + Instant::now(), + ) + .unwrap(); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:12:off", + &surface_success("client-shell-surface:12:off", false, 1), + &mut endpoints, + ); + let old_focus = remote_sent + .lock() + .unwrap() + .iter() + .find_map(|message| match message { + crate::protocol::ClientMessage::ClientShellEndpointRequest { request, .. } + if surface_set_active(message).is_none() => + { + Some( + serde_json::from_str::<crate::api::schema::Request>(request) + .unwrap() + .id, + ) + } + _ => None, + }) + .unwrap(); + let sent_before_retarget = remote_sent.lock().unwrap().len(); + activation + .retarget( + Some(crate::client::shell::ClientEndpointFocusTarget::Workspace( + "new".into(), + )), + &mut endpoints, + ) + .unwrap(); + assert_eq!(remote_sent.lock().unwrap().len(), sent_before_retarget); + assert!(activation.accepts_response(&target, 7, "remote-boot", &old_focus)); + assert_eq!( + activation.receive_response( + &target, + 7, + &old_focus, + &workspace_focus_success(&old_focus, "old"), + &mut endpoints, + ), + SurfaceActivationProgress::Pending + ); + let latest_focus = remote_sent + .lock() + .unwrap() + .last() + .and_then(|message| match message { + crate::protocol::ClientMessage::ClientShellEndpointRequest { request, .. } => Some( + serde_json::from_str::<crate::api::schema::Request>(request) + .unwrap() + .id, + ), + _ => None, + }) + .unwrap(); + assert_ne!(latest_focus, old_focus); + assert!(activation.accepts_response(&target, 7, "remote-boot", &latest_focus)); +} + +#[test] +fn latest_host_focus_is_replayed_to_the_eventual_target() { + let (shell, mut endpoints, _local_sent, remote_sent) = shell_and_registry(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 25, + Instant::now(), + ) + .unwrap(); + activation.update_host_focus(false, &mut endpoints).unwrap(); + assert!(remote_sent.lock().unwrap().is_empty()); + + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:25:off", + &surface_success("client-shell-surface:25:off", false, 1), + &mut endpoints, + ); + assert_eq!( + remote_sent.lock().unwrap().get(2), + Some(&crate::protocol::ClientMessage::ClientShellFocus { focused: false }) + ); + + activation.update_host_focus(true, &mut endpoints).unwrap(); + assert_eq!( + remote_sent.lock().unwrap().last(), + Some(&crate::protocol::ClientMessage::ClientShellFocus { focused: true }) + ); +} + +#[test] +fn host_focus_change_restarts_an_issued_presentation_effects_fence() { + let (_shell, mut endpoints, _local_sent, remote_sent) = shell_and_registry(); + let mut activation = machine(); + activation.phase = ActivationPhase::AwaitingPresentationEffects { + lease: lease(endpoint(), 7, "remote-boot"), + token: "old-token".into(), + ready: false, + completion: Box::new(ActivationCompletion::Activated), + }; + + activation.update_host_focus(false, &mut endpoints).unwrap(); + + assert!(matches!( + activation.phase, + ActivationPhase::SynchronizingPresentation { .. } + )); + assert_eq!( + activation.receive_presentation_effects_ready(&endpoint(), 7, "old-token"), + SurfaceActivationProgress::Stale + ); + assert_eq!( + remote_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![true] + ); +} + +#[test] +fn source_release_rejection_restores_the_source_coherently() { + let (shell, mut endpoints, local_sent, _remote_sent) = shell_and_registry(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 13, + Instant::now(), + ) + .unwrap(); + assert_eq!( + activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:13:off", + &failure("client-shell-surface:13:off", "source rejected release"), + &mut endpoints, + ), + SurfaceActivationProgress::Rejected { + message: "source rejected release".into(), + source_release_rejected: true, + } + ); + assert_eq!( + activation.rollback(&mut endpoints, "source rejected release".into(), true), + ActivationRollback::Pending + ); + assert_eq!(endpoints.active_id(), &ClientEndpointId::Local); + assert!(!endpoints.active_surface_available()); + assert_eq!( + local_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![false, true] + ); +} + +#[test] +fn source_release_timeout_starts_an_acknowledged_source_restore() { + let (shell, mut endpoints, local_sent, _remote_sent) = shell_and_registry(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 14, + Instant::now(), + ) + .unwrap(); + assert_eq!( + activation.rollback(&mut endpoints, "source release timed out".into(), false), + ActivationRollback::Pending + ); + let sent = local_sent.lock().unwrap(); + assert_eq!( + sent.iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![false, true] + ); + assert!(activation.accepts_response( + &ClientEndpointId::Local, + 1, + "local-boot", + "client-shell-surface:14:rollback-source-on" + )); +} + +#[test] +fn resize_invalidates_already_recorded_surface_evidence() { + let (_shell, mut endpoints, _local_sent, _remote_sent) = shell_and_registry(); + let mut activation = machine(); + assert_eq!( + activation.receive_snapshot(&endpoint(), 7, &test_snapshot("remote-boot", 1)), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface(&endpoint(), 7, surface("remote-boot", 1, "pane")), + SurfaceActivationProgress::Ready + ); + let resize = crate::protocol::ClientMessage::ClientShellResize { + cell_width_px: 9, + cell_height_px: 17, + surface_size: crate::protocol::ClientSurfaceSize { + cols: 100, + rows: 30, + }, + pixel_mouse: true, + }; + activation.update_resize(resize, &mut endpoints).unwrap(); + assert_eq!(activation.progress(), SurfaceActivationProgress::Pending); +} + +#[test] +fn resize_during_activation_reaches_the_pending_target() { + let (shell, mut endpoints, _local_sent, remote_sent) = shell_and_registry(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 15, + Instant::now(), + ) + .unwrap(); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:15:off", + &surface_success("client-shell-surface:15:off", false, 1), + &mut endpoints, + ); + let resized = crate::protocol::ClientMessage::ClientShellResize { + cell_width_px: 9, + cell_height_px: 17, + surface_size: crate::protocol::ClientSurfaceSize { + cols: 100, + rows: 30, + }, + pixel_mouse: true, + }; + activation + .update_resize(resized.clone(), &mut endpoints) + .unwrap(); + assert_eq!(remote_sent.lock().unwrap().last(), Some(&resized)); + assert_eq!( + activation.receive_surface(&endpoint(), 7, surface("remote-boot", 1, "pane")), + SurfaceActivationProgress::Pending, + "a surface for the prior geometry cannot commit" + ); +} + +#[test] +fn rapid_a_to_b_to_a_restores_source_before_a_fresh_latest_epoch() { + let (mut shell, mut endpoints, local_sent, remote_sent) = shell_and_registry(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + endpoint(), + None, + resize(), + 20, + Instant::now(), + ) + .unwrap(); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:20:off", + &surface_success("client-shell-surface:20:off", false, 1), + &mut endpoints, + ); + + // The latest A-qualified target replaces B while B may already have accepted target-on. + assert_eq!( + activation.supersede( + ClientEndpointId::Local, + Some(crate::client::shell::ClientEndpointFocusTarget::Pane( + "local-pane".into(), + )), + &mut endpoints, + ), + ActivationRollback::Pending + ); + assert_eq!( + remote_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![true, false], + "B is released before A can be restored" + ); + assert_eq!( + activation.receive_surface(&endpoint(), 7, surface("remote-boot", 2, "pane")), + SurfaceActivationProgress::Stale, + "delayed B activation evidence cannot satisfy A restoration" + ); + let _ = activation.receive_response( + &endpoint(), + 7, + "client-shell-surface:20:rollback-target-off", + &surface_success("client-shell-surface:20:rollback-target-off", false, 1), + &mut endpoints, + ); + assert_eq!( + local_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![false, true], + "source restoration is acknowledged rather than racing target ownership" + ); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:20:rollback-source-on", + &surface_success("client-shell-surface:20:rollback-source-on", true, 2), + &mut endpoints, + ); + let local_snapshot = test_snapshot("local-boot", 2); + shell.set_snapshot(Box::new(local_snapshot.clone())); + assert_eq!( + activation.receive_snapshot(&ClientEndpointId::Local, 1, &local_snapshot), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface( + &ClientEndpointId::Local, + 1, + surface("local-boot", 2, "pane") + ), + SurfaceActivationProgress::Ready + ); + assert!(matches!( + activation.complete(&mut shell, &mut endpoints), + Ok(ActivationCompletion::AwaitingPresentationSync { + endpoint: ClientEndpointId::Local, + .. + }) + )); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:20:presentation-sync", + &surface_success("client-shell-surface:20:presentation-sync", true, 3), + &mut endpoints, + ); + let sync_snapshot = test_snapshot("local-boot", 3); + shell.set_endpoint_snapshot_for_generation( + &ClientEndpointId::Local, + 1, + Box::new(sync_snapshot.clone()), + ); + let _ = activation.receive_snapshot(&ClientEndpointId::Local, 1, &sync_snapshot); + assert_eq!( + activation.receive_surface( + &ClientEndpointId::Local, + 1, + surface("local-boot", 3, "pane") + ), + SurfaceActivationProgress::Ready + ); + assert_eq!( + activation.complete(&mut shell, &mut endpoints), + Ok(ActivationCompletion::AwaitingPresentationEffects) + ); + assert_eq!( + activation.receive_presentation_effects_ready( + &ClientEndpointId::Local, + 1, + "20:1:local-boot" + ), + SurfaceActivationProgress::Ready + ); + assert!(matches!( + activation.complete(&mut shell, &mut endpoints), + Ok(ActivationCompletion::RestoredSource { + successor: Some(EndpointActivationIntent { + endpoint_id: ClientEndpointId::Local, + .. + }), + .. + }) + )); + assert_eq!(endpoints.active_id(), &ClientEndpointId::Local); + + // Runtime queues this successor with force=true, so even source==target receives a new + // activation epoch only after restoration committed. + let _fresh_epoch = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + ClientEndpointId::Local, + Some(crate::client::shell::ClientEndpointFocusTarget::Pane( + "local-pane".into(), + )), + resize(), + 21, + Instant::now(), + ) + .unwrap(); + assert!(local_sent.lock().unwrap().iter().any(|message| { + matches!( + message, + crate::protocol::ClientMessage::ClientShellEndpointRequest { request, .. } + if serde_json::from_str::<crate::api::schema::Request>(request) + .is_ok_and(|request| request.id == "client-shell-surface:21:on") + ) + })); +} + +#[test] +fn disconnected_committed_source_does_not_block_switching_to_a_live_endpoint() { + let (mut shell, mut endpoints, local_sent, _remote_sent) = shell_and_registry(); + let disconnected = endpoint(); + endpoints.set_surface_active(&ClientEndpointId::Local, false); + endpoints.set_surface_active(&disconnected, true); + assert!(endpoints.set_active(&disconnected)); + assert!(shell.activate_endpoint_projection(&disconnected)); + endpoints.disconnect(&disconnected); + + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + ClientEndpointId::Local, + None, + resize(), + 22, + Instant::now(), + ) + .unwrap(); + assert_eq!(activation.source_command_lane(), None); + assert_eq!( + local_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![true], + "there is no unavailable source release to await" + ); + + assert_eq!( + activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:22:on", + &surface_success("client-shell-surface:22:on", true, 2), + &mut endpoints, + ), + SurfaceActivationProgress::Pending + ); + let snapshot = test_snapshot("local-boot", 2); + shell.set_endpoint_snapshot_for_generation( + &ClientEndpointId::Local, + 1, + Box::new(snapshot.clone()), + ); + assert_eq!( + activation.receive_snapshot(&ClientEndpointId::Local, 1, &snapshot), + SurfaceActivationProgress::Pending + ); + assert_eq!( + activation.receive_surface( + &ClientEndpointId::Local, + 1, + surface("local-boot", 2, "pane") + ), + SurfaceActivationProgress::Ready + ); + assert!(matches!( + activation.complete(&mut shell, &mut endpoints), + Ok(ActivationCompletion::AwaitingPresentationSync { + previous, + endpoint: ClientEndpointId::Local, + }) if previous == disconnected + )); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:22:presentation-sync", + &surface_success("client-shell-surface:22:presentation-sync", true, 3), + &mut endpoints, + ); + let sync_snapshot = test_snapshot("local-boot", 3); + shell.set_endpoint_snapshot_for_generation( + &ClientEndpointId::Local, + 1, + Box::new(sync_snapshot.clone()), + ); + let _ = activation.receive_snapshot(&ClientEndpointId::Local, 1, &sync_snapshot); + assert_eq!( + activation.receive_surface( + &ClientEndpointId::Local, + 1, + surface("local-boot", 3, "pane") + ), + SurfaceActivationProgress::Ready + ); + assert_eq!( + activation.complete(&mut shell, &mut endpoints), + Ok(ActivationCompletion::AwaitingPresentationEffects) + ); + assert_eq!( + activation.receive_presentation_effects_ready( + &ClientEndpointId::Local, + 1, + "22:1:local-boot" + ), + SurfaceActivationProgress::Ready + ); + assert_eq!( + activation.complete(&mut shell, &mut endpoints), + Ok(ActivationCompletion::Activated) + ); + assert_eq!(endpoints.active_id(), &ClientEndpointId::Local); + assert_ne!(endpoints.active_id(), &disconnected); +} + +#[test] +fn rollback_keeps_the_latest_intent_even_when_it_returns_to_the_target() { + let (shell, mut endpoints, _local_sent, _remote_sent) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + None, + resize(), + 23, + Instant::now(), + ) + .unwrap(); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:23:off", + &surface_success("client-shell-surface:23:off", false, 1), + &mut endpoints, + ); + assert_eq!( + activation.supersede( + ClientEndpointId::Local, + Some(crate::client::shell::ClientEndpointFocusTarget::Pane( + "local-pane".into() + )), + &mut endpoints, + ), + ActivationRollback::Pending + ); + assert!(!activation.can_retarget(&target)); + assert_eq!( + activation.supersede( + target.clone(), + Some(crate::client::shell::ClientEndpointFocusTarget::Pane( + "remote-pane".into() + )), + &mut endpoints, + ), + ActivationRollback::Pending + ); + assert_eq!( + activation.successor, + Some(EndpointActivationIntent { + endpoint_id: target, + target: Some(crate::client::shell::ClientEndpointFocusTarget::Pane( + "remote-pane".into() + )), + }) + ); +} + +#[test] +fn unacknowledged_target_release_closes_target_before_restoring_source() { + let (shell, mut endpoints, local_sent, _remote_sent) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + None, + resize(), + 24, + Instant::now(), + ) + .unwrap(); + let _ = activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:24:off", + &surface_success("client-shell-surface:24:off", false, 1), + &mut endpoints, + ); + assert_eq!( + activation.rollback(&mut endpoints, "target activation timed out".into(), false), + ActivationRollback::Pending + ); + assert_eq!( + activation.rollback(&mut endpoints, "target release timed out".into(), false), + ActivationRollback::Pending + ); + assert!(endpoints.connection(&target).is_none()); + let failures = endpoints.take_failures(); + assert_eq!( + failures.len(), + 1, + "rollback revocation must reach the reconnect owner" + ); + assert_eq!(failures[0].endpoint_id, target); + assert_eq!(failures[0].generation, 7); + assert_eq!(failures[0].kind, std::io::ErrorKind::TimedOut); + assert_eq!( + local_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![false, true] + ); +} + +#[test] +fn target_loss_at_activation_deadline_restores_source_before_timeout() { + let (shell, mut endpoints, local_sent, _) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + None, + resize(), + 30, + Instant::now(), + ) + .unwrap(); + activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:30:off", + &surface_success("client-shell-surface:30:off", false, 1), + &mut endpoints, + ); + let now = Instant::now(); + activation.deadline = now; + assert!(activation.expired(now)); + endpoints.fail(&target, std::io::ErrorKind::UnexpectedEof.into()); + // Match the client timer: apply transport failures before checking phase expiry. + for failure in endpoints.take_failures() { + assert_eq!( + activation.endpoint_disconnected(&mut endpoints, &failure.endpoint_id, failure.message), + ActivationRollback::Pending + ); + } + assert!(matches!( + activation.phase, + ActivationPhase::RestoringSource { .. } + )); + assert!(!activation.expired(now)); + assert_eq!( + local_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![false, true] + ); +} + +#[test] +fn losing_local_during_handoff_does_not_revoke_the_healthy_target() { + for source_released in [false, true] { + let (shell, mut endpoints, _local_sent, remote_sent) = shell_and_registry(); + let target = endpoint(); + let mut activation = PendingEndpointActivation::begin( + &shell, + &mut endpoints, + target.clone(), + None, + resize(), + 29, + Instant::now(), + ) + .unwrap(); + if source_released { + activation.receive_response( + &ClientEndpointId::Local, + 1, + "client-shell-surface:29:off", + &surface_success("client-shell-surface:29:off", false, 1), + &mut endpoints, + ); + } + if !source_released { + activation.deadline = Instant::now() - Duration::from_millis(1); + } + endpoints.fail( + &ClientEndpointId::Local, + std::io::ErrorKind::BrokenPipe.into(), + ); + assert_eq!( + activation.endpoint_disconnected( + &mut endpoints, + &ClientEndpointId::Local, + "Local stopped".into() + ), + ActivationRollback::Pending + ); + assert!(!activation.source_available); + assert!( + !activation.expired(Instant::now()), + "starting the healthy target must get a fresh deadline" + ); + assert!(matches!( + activation.phase, + ActivationPhase::ActivatingTarget { .. } + )); + assert!(endpoints.connection(&target).is_some()); + assert_eq!( + remote_sent + .lock() + .unwrap() + .iter() + .filter_map(surface_set_active) + .collect::<Vec<_>>(), + vec![true] + ); + } +} + +#[test] +fn resize_message_preserves_the_latest_surface_dimensions() { + assert_eq!( + resize_geometry(&resize()), + Some(crate::protocol::ClientSurfaceSize { cols: 80, rows: 24 }) + ); +} diff --git a/src/client/endpoint/catalog.rs b/src/client/endpoint/catalog.rs new file mode 100644 index 00000000..23f26387 --- /dev/null +++ b/src/client/endpoint/catalog.rs @@ -0,0 +1,578 @@ +use std::collections::HashSet; +use std::io::{self, Read as _, Write as _}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::{Deserialize, Serialize}; + +use super::ProfileId; + +const CATALOG_VERSION: u32 = 1; +const SELECTION_VERSION: u32 = 1; +const MAX_CATALOG_BYTES: u64 = 64 * 1024; +const MAX_PROFILES: usize = 64; +const MAX_LABEL_BYTES: usize = 128; +const MAX_TARGET_BYTES: usize = 1024; +static NEXT_TEMP_FILE: AtomicU64 = AtomicU64::new(1); + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct SavedSshEndpoint { + pub(crate) id: ProfileId, + pub(crate) label: String, + pub(crate) target: String, + pub(crate) session: String, + pub(crate) enabled: bool, +} + +impl SavedSshEndpoint { + pub(crate) fn new( + label: impl Into<String>, + target: impl Into<String>, + session: impl Into<String>, + ) -> Result<Self, String> { + let profile = Self { + id: ProfileId::generate(), + label: label.into(), + target: target.into(), + session: session.into(), + enabled: true, + }; + profile.validate()?; + Ok(profile) + } + + fn validate(&self) -> Result<(), String> { + ProfileId::parse(self.id.to_string())?; + let label = self.label.trim(); + if label.is_empty() { + return Err("SSH endpoint label cannot be empty".into()); + } + if label.len() > MAX_LABEL_BYTES || label.chars().any(char::is_control) { + return Err(format!( + "SSH endpoint label must be at most {MAX_LABEL_BYTES} bytes and contain no control characters" + )); + } + if self.target.len() > MAX_TARGET_BYTES || self.target.chars().any(char::is_control) { + return Err(format!( + "SSH target must be at most {MAX_TARGET_BYTES} bytes and contain no control characters" + )); + } + crate::remote::validate_remote_target(&self.target).map(|_| ())?; + let authority = self.target.strip_prefix("ssh://").unwrap_or(&self.target); + if authority + .rsplit_once('@') + .is_some_and(|(userinfo, _)| userinfo.contains(':')) + { + return Err("SSH target must not contain a password".into()); + } + crate::session::validate_name(&self.session)?; + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct EndpointCatalog { + version: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) selected_profile: Option<ProfileId>, + #[serde(default)] + pub(crate) ssh: Vec<SavedSshEndpoint>, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct EndpointSelection { + version: u32, + selected_profile: Option<ProfileId>, +} + +impl Default for EndpointCatalog { + fn default() -> Self { + Self { + version: CATALOG_VERSION, + selected_profile: None, + ssh: Vec::new(), + } + } +} + +impl EndpointCatalog { + pub(crate) fn load() -> Result<Self, String> { + Self::load_from_paths(&catalog_path(), &selection_path()) + } + + pub(crate) fn load_profiles() -> Result<Vec<SavedSshEndpoint>, String> { + // Live clients keep their own selection, independent of other attached clients. + Self::load_from_path(&catalog_path()).map(|catalog| catalog.ssh) + } + + fn load_from_paths(catalog_path: &Path, selection_path: &Path) -> Result<Self, String> { + let mut catalog = Self::load_from_path(catalog_path)?; + match load_selection_from_path(selection_path) { + Ok(Some(selection)) => { + let valid = selection.selected_profile.as_ref().is_none_or(|selected| { + catalog + .ssh + .iter() + .any(|profile| &profile.id == selected && profile.enabled) + }); + if valid { + catalog.selected_profile = selection.selected_profile; + } else { + tracing::warn!( + path = %selection_path.display(), + "saved endpoint selection is absent or disabled; using Local" + ); + } + } + Ok(None) => {} + Err(error) => { + tracing::warn!( + %error, + path = %selection_path.display(), + "saved endpoint selection is unavailable; using Local" + ); + } + } + Ok(catalog) + } + + pub(crate) fn store_profiles(&self) -> Result<(), String> { + self.store_to_path(&catalog_path()) + } + + pub(crate) fn store_selection(&self) -> Result<(), String> { + self.store_selection_to_path(&selection_path()) + } + + fn store_selection_to_path(&self, path: &Path) -> Result<(), String> { + self.validate()?; + let content = serde_json::to_vec_pretty(&EndpointSelection { + version: SELECTION_VERSION, + selected_profile: self.selected_profile.clone(), + }) + .map_err(|error| format!("failed to encode endpoint selection: {error}"))?; + store_private_json(path, &content, "endpoint selection") + } + + pub(crate) fn add_ssh( + &mut self, + label: impl Into<String>, + target: impl Into<String>, + session: impl Into<String>, + ) -> Result<ProfileId, String> { + if self.ssh.len() >= MAX_PROFILES { + return Err(format!("at most {MAX_PROFILES} SSH endpoints can be saved")); + } + let profile = SavedSshEndpoint::new(label, target, session)?; + let id = profile.id.clone(); + self.ssh.push(profile); + Ok(id) + } + + pub(crate) fn rename_ssh( + &mut self, + id: &ProfileId, + label: impl Into<String>, + ) -> Result<bool, String> { + let Some(index) = self.ssh.iter().position(|profile| &profile.id == id) else { + return Ok(false); + }; + let mut renamed = self.ssh[index].clone(); + renamed.label = label.into(); + renamed.validate()?; + self.ssh[index] = renamed; + Ok(true) + } + + pub(crate) fn remove_ssh(&mut self, id: &ProfileId) -> bool { + let previous_len = self.ssh.len(); + self.ssh.retain(|profile| &profile.id != id); + if self.selected_profile.as_ref() == Some(id) { + self.selected_profile = None; + } + self.ssh.len() != previous_len + } + + pub(crate) fn select_local(&mut self) { + self.selected_profile = None; + } + + pub(crate) fn select_endpoint(&mut self, endpoint_id: &super::ClientEndpointId) -> bool { + match endpoint_id { + super::ClientEndpointId::Local => { + self.select_local(); + true + } + super::ClientEndpointId::Ssh(profile_id) => self.select_ssh(profile_id), + } + } + + pub(crate) fn select_ssh(&mut self, id: &ProfileId) -> bool { + if !self + .ssh + .iter() + .any(|profile| &profile.id == id && profile.enabled) + { + return false; + } + self.selected_profile = Some(id.clone()); + true + } + + pub(crate) fn has_enabled_ssh(&self) -> bool { + self.ssh.iter().any(|profile| profile.enabled) + } + + pub(crate) fn contains_enabled_target_session(&self, target: &str, session: &str) -> bool { + self.ssh.iter().any(|profile| { + profile.enabled && profile.target == target && profile.session == session + }) + } + + pub(crate) fn set_enabled(&mut self, id: &ProfileId, enabled: bool) -> bool { + let Some(profile) = self.ssh.iter_mut().find(|profile| &profile.id == id) else { + return false; + }; + profile.enabled = enabled; + if !enabled && self.selected_profile.as_ref() == Some(id) { + self.selected_profile = None; + } + true + } + + fn validate(&self) -> Result<(), String> { + if self.version != CATALOG_VERSION { + return Err(format!( + "unsupported endpoint catalog version {}; expected {CATALOG_VERSION}", + self.version + )); + } + if self.ssh.len() > MAX_PROFILES { + return Err(format!( + "endpoint catalog contains more than {MAX_PROFILES} SSH profiles" + )); + } + let mut ids = HashSet::new(); + for profile in &self.ssh { + profile.validate()?; + if !ids.insert(profile.id.clone()) { + return Err(format!("duplicate endpoint profile id {}", profile.id)); + } + } + if self.selected_profile.as_ref().is_some_and(|selected| { + !self + .ssh + .iter() + .any(|profile| &profile.id == selected && profile.enabled) + }) { + return Err("selected SSH endpoint is absent or disabled in the catalog".into()); + } + Ok(()) + } + + fn load_from_path(path: &Path) -> Result<Self, String> { + let file = match std::fs::File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Self::default()), + Err(error) => { + return Err(format!( + "failed to open endpoint catalog {}: {error}", + path.display() + )) + } + }; + let metadata = file + .metadata() + .map_err(|error| format!("failed to inspect endpoint catalog: {error}"))?; + if metadata.len() > MAX_CATALOG_BYTES { + return Err("endpoint catalog exceeds the storage limit".into()); + } + let mut content = String::new(); + file.take(MAX_CATALOG_BYTES + 1) + .read_to_string(&mut content) + .map_err(|error| format!("failed to read endpoint catalog: {error}"))?; + if content.len() as u64 > MAX_CATALOG_BYTES { + return Err("endpoint catalog exceeds the storage limit".into()); + } + let catalog: Self = serde_json::from_str(&content) + .map_err(|error| format!("stored endpoint catalog is invalid: {error}"))?; + catalog.validate()?; + Ok(catalog) + } + + fn store_to_path(&self, path: &Path) -> Result<(), String> { + self.validate()?; + let content = serde_json::to_vec_pretty(self) + .map_err(|error| format!("failed to encode endpoint catalog: {error}"))?; + store_private_json(path, &content, "endpoint catalog") + } +} + +fn load_selection_from_path(path: &Path) -> Result<Option<EndpointSelection>, String> { + let content = match std::fs::read(path) { + Ok(content) => content, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(format!("failed to read endpoint selection: {error}")), + }; + if content.len() as u64 > MAX_CATALOG_BYTES { + return Err("endpoint selection exceeds the storage limit".into()); + } + let selection: EndpointSelection = serde_json::from_slice(&content) + .map_err(|error| format!("stored endpoint selection is invalid: {error}"))?; + if selection.version != SELECTION_VERSION { + return Err(format!( + "unsupported endpoint selection version {}; expected {SELECTION_VERSION}", + selection.version + )); + } + Ok(Some(selection)) +} + +fn store_private_json(path: &Path, content: &[u8], description: &str) -> Result<(), String> { + if content.len() as u64 > MAX_CATALOG_BYTES { + return Err(format!("{description} exceeds the storage limit")); + } + let parent = path + .parent() + .ok_or_else(|| format!("invalid {description} path: {}", path.display()))?; + std::fs::create_dir_all(parent) + .map_err(|error| format!("failed to create {description} directory: {error}"))?; + if let Ok(metadata) = std::fs::symlink_metadata(path) { + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(format!( + "refusing to replace {description} through a non-file path" + )); + } + } + + let sequence = NEXT_TEMP_FILE.fetch_add(1, Ordering::Relaxed); + let temp_path = parent.join(format!(".endpoints-{}-{sequence}.tmp", std::process::id())); + let mut temp = crate::platform::create_private_state_file(&temp_path) + .map_err(|error| format!("failed to create {description}: {error}"))?; + if let Err(error) = temp.write_all(content).and_then(|()| temp.sync_all()) { + drop(temp); + let _ = std::fs::remove_file(&temp_path); + return Err(format!("failed to write {description}: {error}")); + } + drop(temp); + if let Err(error) = crate::platform::replace_file(&temp_path, path) { + let _ = std::fs::remove_file(&temp_path); + return Err(format!("failed to activate {description}: {error}")); + } + crate::platform::sync_parent_directory(parent) + .map_err(|error| format!("failed to persist {description} directory: {error}")) +} + +pub(crate) fn catalog_path() -> PathBuf { + crate::config::state_dir() + .join("client") + .join("endpoints.json") +} + +fn selection_path() -> PathBuf { + crate::config::state_dir() + .join("client") + .join("endpoint-selection.json") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn path(name: &str) -> PathBuf { + std::env::temp_dir() + .join(format!( + "herdr-endpoint-catalog-{}-{name}", + std::process::id() + )) + .join("endpoints.json") + } + + #[test] + fn catalog_roundtrip_persists_profiles_without_secret_fields() { + let path = path("roundtrip"); + let _ = std::fs::remove_dir_all(path.parent().unwrap()); + let mut catalog = EndpointCatalog::default(); + let id = catalog + .add_ssh("Build", "ssh://dev@build.example:2222", "agents") + .unwrap(); + assert!(catalog.select_ssh(&id)); + catalog.store_to_path(&path).unwrap(); + + let encoded = std::fs::read_to_string(&path).unwrap(); + assert!(!encoded.contains("password")); + assert!(!encoded.contains("private_key")); + assert!(!encoded.contains("control_socket")); + let loaded = EndpointCatalog::load_from_path(&path).unwrap(); + assert_eq!(loaded, catalog); + assert_eq!(loaded.ssh[0].id, id); + std::fs::remove_dir_all(path.parent().unwrap()).unwrap(); + } + + #[test] + fn duplicate_target_and_session_profiles_keep_distinct_opaque_ids() { + let mut catalog = EndpointCatalog::default(); + let first = catalog.add_ssh("One", "build", "default").unwrap(); + let second = catalog.add_ssh("Two", "build", "default").unwrap(); + assert_ne!(first, second); + } + + #[test] + fn catalog_rejects_passwords_embedded_in_ssh_targets() { + let mut catalog = EndpointCatalog::default(); + assert!(catalog + .add_ssh("Build", "ssh://dev:secret@build.example", "default") + .unwrap_err() + .contains("must not contain a password")); + assert!(catalog + .add_ssh("Build", "dev:secret@build.example", "default") + .is_err()); + assert!(catalog + .add_ssh("Build", "ssh://dev@[::1]:2222", "default") + .is_ok()); + } + + #[test] + fn interactive_bootstrap_matches_only_enabled_target_and_session() { + let mut catalog = EndpointCatalog::default(); + let id = catalog.add_ssh("Build", "build", "agents").unwrap(); + assert!(catalog.contains_enabled_target_session("build", "agents")); + assert!(!catalog.contains_enabled_target_session("build", "default")); + assert!(catalog.set_enabled(&id, false)); + assert!(!catalog.contains_enabled_target_session("build", "agents")); + } + + #[test] + fn rename_changes_only_the_machine_label() { + let mut catalog = EndpointCatalog::default(); + let id = catalog.add_ssh("Old", "build", "agents").unwrap(); + let original = catalog.ssh[0].clone(); + + assert!(catalog.rename_ssh(&id, "New").unwrap()); + assert_eq!(catalog.ssh[0].label, "New"); + assert_eq!(catalog.ssh[0].id, original.id); + assert_eq!(catalog.ssh[0].target, original.target); + assert_eq!(catalog.ssh[0].session, original.session); + assert!(catalog.rename_ssh(&id, "\n").is_err()); + assert_eq!(catalog.ssh[0].label, "New"); + } + + #[test] + fn removal_and_disable_return_selection_to_local() { + let mut catalog = EndpointCatalog::default(); + let first = catalog.add_ssh("One", "one", "default").unwrap(); + assert!(catalog.select_ssh(&first)); + assert!(catalog.set_enabled(&first, false)); + assert_eq!(catalog.selected_profile, None); + + assert!(catalog.set_enabled(&first, true)); + assert!(catalog.select_ssh(&first)); + assert!(catalog.remove_ssh(&first)); + assert_eq!(catalog.selected_profile, None); + } + + #[test] + fn catalog_rejects_unknown_fields_instead_of_retaining_possible_secrets() { + let path = path("unknown-field"); + let _ = std::fs::remove_dir_all(path.parent().unwrap()); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write( + &path, + r#"{ + "version": 1, + "ssh": [{ + "id": "0123456789abcdef0123456789abcdef", + "label": "Build", + "target": "build", + "session": "default", + "enabled": true, + "password": "must-not-be-accepted" + }] + }"#, + ) + .unwrap(); + assert!(EndpointCatalog::load_from_path(&path) + .unwrap_err() + .contains("unknown field")); + std::fs::remove_dir_all(path.parent().unwrap()).unwrap(); + } + + #[test] + fn storing_selection_does_not_rewrite_profile_membership() { + let catalog_path = path("separate-selection"); + let selection_path = catalog_path.with_file_name("selection.json"); + let _ = std::fs::remove_dir_all(catalog_path.parent().unwrap()); + let mut catalog = EndpointCatalog::default(); + let id = catalog.add_ssh("Build", "build", "agents").unwrap(); + catalog.store_to_path(&catalog_path).unwrap(); + let profiles_before = std::fs::read(&catalog_path).unwrap(); + + assert!(catalog.select_ssh(&id)); + catalog.store_selection_to_path(&selection_path).unwrap(); + + assert_eq!(std::fs::read(&catalog_path).unwrap(), profiles_before); + assert_eq!( + load_selection_from_path(&selection_path) + .unwrap() + .unwrap() + .selected_profile, + Some(id) + ); + std::fs::remove_dir_all(catalog_path.parent().unwrap()).unwrap(); + } + + #[test] + fn malformed_selection_does_not_discard_saved_profiles() { + let catalog_path = path("malformed-selection"); + let selection_path = catalog_path.with_file_name("selection.json"); + let _ = std::fs::remove_dir_all(catalog_path.parent().unwrap()); + let mut catalog = EndpointCatalog::default(); + let id = catalog.add_ssh("Build", "build", "agents").unwrap(); + catalog.store_to_path(&catalog_path).unwrap(); + std::fs::write(&selection_path, b"not json").unwrap(); + + let loaded = EndpointCatalog::load_from_paths(&catalog_path, &selection_path).unwrap(); + assert_eq!(loaded.ssh.len(), 1); + assert_eq!(loaded.ssh[0].id, id); + assert_eq!(loaded.selected_profile, None); + std::fs::remove_dir_all(catalog_path.parent().unwrap()).unwrap(); + } + + #[test] + fn absent_selected_profile_falls_back_without_discarding_catalog() { + let catalog_path = path("absent-selection"); + let selection_path = catalog_path.with_file_name("selection.json"); + let _ = std::fs::remove_dir_all(catalog_path.parent().unwrap()); + let mut catalog = EndpointCatalog::default(); + let saved = catalog.add_ssh("Build", "build", "agents").unwrap(); + catalog.store_to_path(&catalog_path).unwrap(); + let missing = ProfileId::parse("fedcba9876543210fedcba9876543210").unwrap(); + store_private_json( + &selection_path, + &serde_json::to_vec(&EndpointSelection { + version: SELECTION_VERSION, + selected_profile: Some(missing), + }) + .unwrap(), + "endpoint selection", + ) + .unwrap(); + + let loaded = EndpointCatalog::load_from_paths(&catalog_path, &selection_path).unwrap(); + assert_eq!(loaded.ssh[0].id, saved); + assert_eq!(loaded.selected_profile, None); + std::fs::remove_dir_all(catalog_path.parent().unwrap()).unwrap(); + } + + #[test] + fn invalid_or_missing_selected_profile_is_rejected() { + let catalog = EndpointCatalog { + selected_profile: Some(ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap()), + ..EndpointCatalog::default() + }; + assert!(catalog.validate().is_err()); + } +} diff --git a/src/client/endpoint/control.rs b/src/client/endpoint/control.rs new file mode 100644 index 00000000..7cf1159f --- /dev/null +++ b/src/client/endpoint/control.rs @@ -0,0 +1,63 @@ +use super::ClientEndpointId; + +pub(crate) enum EndpointControlMessage { + HealthPong, + Snapshot(Box<crate::protocol::ClientShellSnapshot>), + Ignored, +} + +pub(crate) fn decode_endpoint_control( + kind: &str, + data: &str, +) -> Result<EndpointControlMessage, String> { + if kind == crate::protocol::endpoint::HEALTH_PONG_KIND { + return Ok(EndpointControlMessage::HealthPong); + } + if kind == crate::protocol::endpoint::ENDPOINT_SNAPSHOT_KIND { + let snapshot = serde_json::from_str(data) + .map_err(|error| format!("invalid endpoint snapshot: {error}"))?; + return Ok(EndpointControlMessage::Snapshot(Box::new(snapshot))); + } + if kind.starts_with("shell.snapshot.") { + return Err(format!( + "unsupported mandatory endpoint snapshot codec {kind:?}" + )); + } + Ok(EndpointControlMessage::Ignored) +} + +pub(crate) fn protocol_failure_is_fatal(endpoint_id: &ClientEndpointId) -> bool { + endpoint_id.is_local() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::client::endpoint::ProfileId; + + #[test] + fn unknown_optional_controls_are_ignored() { + assert!(matches!( + decode_endpoint_control("future.optional", "not json").unwrap(), + EndpointControlMessage::Ignored + )); + } + + #[test] + fn unknown_snapshot_codecs_are_rejected() { + assert_eq!( + decode_endpoint_control("shell.snapshot.v2", "{}") + .err() + .as_deref(), + Some("unsupported mandatory endpoint snapshot codec \"shell.snapshot.v2\"") + ); + } + + #[test] + fn only_local_protocol_failures_end_the_client() { + let remote = + ClientEndpointId::Ssh(ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap()); + assert!(protocol_failure_is_fatal(&ClientEndpointId::Local)); + assert!(!protocol_failure_is_fatal(&remote)); + } +} diff --git a/src/client/endpoint/health.rs b/src/client/endpoint/health.rs new file mode 100644 index 00000000..87c08952 --- /dev/null +++ b/src/client/endpoint/health.rs @@ -0,0 +1,99 @@ +use std::time::{Duration, Instant}; + +pub(super) const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); +pub(super) const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(10); + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum HealthAction { + None, + Ping, + Expired, +} + +pub(super) struct EndpointHealth { + connected_at: Instant, + last_received: Instant, + ping_sent_at: Option<Instant>, + ready: bool, +} + +impl EndpointHealth { + pub(super) fn new(now: Instant) -> Self { + Self { + connected_at: now, + last_received: now, + ping_sent_at: None, + ready: false, + } + } + + pub(super) fn received(&mut self, now: Instant) { + self.last_received = now; + self.ping_sent_at = None; + } + + pub(super) fn ready(&mut self) { + self.ready = true; + } + + pub(super) fn action(&self, now: Instant) -> HealthAction { + let initial_snapshot_expired = + !self.ready && now.saturating_duration_since(self.connected_at) >= HEARTBEAT_TIMEOUT; + let probe_expired = self + .ping_sent_at + .is_some_and(|sent_at| now.saturating_duration_since(sent_at) >= HEARTBEAT_TIMEOUT); + if initial_snapshot_expired || probe_expired { + HealthAction::Expired + } else if self.ping_sent_at.is_none() + && now.saturating_duration_since(self.last_received) >= HEARTBEAT_INTERVAL + { + HealthAction::Ping + } else { + HealthAction::None + } + } + + pub(super) fn ping_sent(&mut self, now: Instant) { + self.ping_sent_at = Some(now); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn quiet_connection_is_probed_then_expires_without_a_reply() { + let now = Instant::now(); + let mut health = EndpointHealth::new(now); + assert_eq!(health.action(now), HealthAction::None); + assert_eq!(health.action(now + HEARTBEAT_INTERVAL), HealthAction::Ping); + health.ping_sent(now + HEARTBEAT_INTERVAL); + assert_eq!( + health.action(now + HEARTBEAT_INTERVAL + HEARTBEAT_TIMEOUT), + HealthAction::Expired + ); + } + + #[test] + fn any_incoming_message_satisfies_an_outstanding_probe() { + let now = Instant::now(); + let mut health = EndpointHealth::new(now); + health.ready(); + health.ping_sent(now); + health.received(now + HEARTBEAT_TIMEOUT - Duration::from_millis(1)); + assert_eq!(health.action(now + HEARTBEAT_TIMEOUT), HealthAction::None); + } + + #[test] + fn heartbeats_do_not_hide_a_missing_initial_snapshot() { + let now = Instant::now(); + let mut health = EndpointHealth::new(now); + health.ping_sent(now + HEARTBEAT_INTERVAL); + health.received(now + HEARTBEAT_INTERVAL + Duration::from_secs(1)); + assert_eq!( + health.action(now + HEARTBEAT_TIMEOUT), + HealthAction::Expired + ); + } +} diff --git a/src/client/endpoint/message_policy.rs b/src/client/endpoint/message_policy.rs new file mode 100644 index 00000000..5e84325a --- /dev/null +++ b/src/client/endpoint/message_policy.rs @@ -0,0 +1,124 @@ +use crate::protocol::ServerMessage; + +/// Effects that are meaningful only for the endpoint currently holding the host presentation +/// lease. During a frozen handoff they are intentionally dropped and the committed target asks +/// for one bounded replay, rather than allowing target graphics or modes onto the source frame. +pub(crate) fn is_presentation_effect(message: &ServerMessage) -> bool { + matches!( + message, + ServerMessage::MouseCapture { .. } + | ServerMessage::ClientShellKeyboardReportAll { .. } + | ServerMessage::WindowTitle { .. } + | ServerMessage::Graphics { .. } + | ServerMessage::GraphicsFile { .. } + | ServerMessage::GraphicsTransmissionRetired { .. } + | ServerMessage::TerminalBell { .. } + ) +} + +pub(crate) fn accepts_endpoint_message( + endpoint_active: bool, + activation_message: bool, + command_response: bool, + message: &ServerMessage, +) -> bool { + endpoint_active + || matches!( + message, + ServerMessage::EndpointControl { .. } + | ServerMessage::SemanticNotification(_) + | ServerMessage::ServerShutdown { .. } + | ServerMessage::ClientShellSnapshot(_) + | ServerMessage::Welcome { .. } + ) + || (activation_message + && matches!( + message, + ServerMessage::PaneSurface(_) + | ServerMessage::ClientShellEndpointResponseChunk { .. } + )) + || (command_response + && matches!( + message, + ServerMessage::ClientShellEndpointResponseChunk { .. } + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn inactive_endpoints_deliver_metadata_but_not_presentation_effects() { + assert!(accepts_endpoint_message( + false, + false, + false, + &ServerMessage::EndpointControl { + kind: "snapshot".into(), + data: "{}".into(), + } + )); + assert!(!accepts_endpoint_message( + false, + false, + false, + &ServerMessage::Clipboard { + data: "text".into() + } + )); + assert!(!accepts_endpoint_message( + false, + false, + false, + &ServerMessage::WindowTitle { + title: Some("remote".into()) + } + )); + } + + #[test] + fn frozen_handoffs_drop_target_effects_until_the_post_commit_resync() { + assert!(is_presentation_effect(&ServerMessage::MouseCapture { + enabled: true, + sgr_pixels: false, + })); + assert!(is_presentation_effect(&ServerMessage::GraphicsFile { + path: "asset".into(), + expected_len: 1, + image_id: 1, + transfer_id: 1, + leading: Vec::new(), + control: "a=t".into(), + surface_asset: None, + })); + assert!(!is_presentation_effect(&ServerMessage::EndpointControl { + kind: "shell.snapshot.v1".into(), + data: "{}".into(), + })); + } + + #[test] + fn pending_activation_accepts_only_its_surface_lane() { + assert!(accepts_endpoint_message( + false, + true, + false, + &ServerMessage::ClientShellEndpointResponseChunk { + boot_id: "boot".into(), + request_id: "surface".into(), + final_chunk: true, + data: Vec::new(), + } + )); + assert!(!accepts_endpoint_message( + false, + true, + false, + &ServerMessage::MouseCapture { + enabled: true, + sgr_pixels: false, + } + )); + } +} diff --git a/src/client/endpoint/registry.rs b/src/client/endpoint/registry.rs new file mode 100644 index 00000000..437f39f3 --- /dev/null +++ b/src/client/endpoint/registry.rs @@ -0,0 +1,629 @@ +use std::collections::{HashMap, HashSet}; +use std::io; +use std::time::Instant; + +use super::health::{EndpointHealth, HealthAction}; +use super::ClientEndpointId; +use crate::protocol::ClientMessage; + +pub(crate) trait EndpointTransport: Send { + fn send(&mut self, message: &ClientMessage) -> io::Result<()>; + + fn disconnect(&mut self) {} + + fn flush(&mut self, _deadline: Instant) -> io::Result<()> { + Ok(()) + } + + fn take_error(&mut self) -> Option<io::Error> { + None + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(crate) struct EndpointNegotiation { + methods: HashSet<String>, + capabilities: HashSet<String>, +} + +impl EndpointNegotiation { + pub(crate) fn new(methods: Vec<String>, capabilities: Vec<String>) -> Self { + Self { + methods: methods.into_iter().collect(), + capabilities: capabilities.into_iter().collect(), + } + } + + pub(crate) fn methods(&self) -> Vec<String> { + self.methods.iter().cloned().collect() + } + + pub(crate) fn supports_method(&self, method: &str) -> bool { + self.methods.contains(method) + } + + pub(crate) fn supports_capability(&self, capability: &str) -> bool { + self.capabilities.contains(capability) + } + + pub(crate) fn supports_surface_interest(&self) -> bool { + self.supports_capability(crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY) + && self.supports_capability( + crate::protocol::endpoint::PRESENTATION_EFFECTS_FENCE_CAPABILITY, + ) + && self.supports_method("client_shell.surface.set") + } + + pub(crate) fn supports_health_check(&self) -> bool { + self.supports_capability(crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY) + } +} + +pub(crate) struct EndpointConnection { + transport: Box<dyn EndpointTransport>, + pub(crate) generation: u64, + pub(crate) surface_active: bool, + pub(crate) negotiation: EndpointNegotiation, + health: Option<EndpointHealth>, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct EndpointTransportFailure { + pub(crate) endpoint_id: ClientEndpointId, + pub(crate) generation: u64, + pub(crate) kind: io::ErrorKind, + pub(crate) message: String, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum EndpointSendOutcome { + Sent, + NotSent, +} + +pub(crate) struct EndpointRegistry { + active: ClientEndpointId, + input_enabled: bool, + connections: HashMap<ClientEndpointId, EndpointConnection>, + failures: Vec<EndpointTransportFailure>, +} + +impl EndpointRegistry { + pub(crate) fn empty() -> Self { + Self { + active: ClientEndpointId::Local, + input_enabled: false, + connections: HashMap::new(), + failures: Vec::new(), + } + } + + pub(crate) fn new( + local: impl EndpointTransport + 'static, + generation: u64, + negotiation: EndpointNegotiation, + ) -> Self { + let mut registry = Self::empty(); + registry.input_enabled = true; + registry.insert( + ClientEndpointId::Local, + local, + generation, + negotiation, + true, + ); + registry + } + + pub(crate) fn active_id(&self) -> &ClientEndpointId { + &self.active + } + + pub(crate) fn active_surface_available(&self) -> bool { + self.input_enabled + && self + .connections + .get(&self.active) + .is_some_and(|connection| connection.surface_active) + } + + pub(crate) fn select_unavailable_local(&mut self) { + self.active = ClientEndpointId::Local; + self.freeze_input(); + } + + pub(crate) fn freeze_input(&mut self) { + self.input_enabled = false; + } + + pub(crate) fn unfreeze_input(&mut self) { + self.input_enabled = true; + } + + pub(crate) fn connection(&self, endpoint_id: &ClientEndpointId) -> Option<&EndpointConnection> { + self.connections.get(endpoint_id) + } + + pub(crate) fn insert( + &mut self, + endpoint_id: ClientEndpointId, + transport: impl EndpointTransport + 'static, + generation: u64, + negotiation: EndpointNegotiation, + surface_active: bool, + ) { + let health = (!endpoint_id.is_local() && negotiation.supports_health_check()) + .then(|| EndpointHealth::new(Instant::now())); + if let Some(mut previous) = self.connections.insert( + endpoint_id, + EndpointConnection { + transport: Box::new(transport), + generation, + surface_active, + negotiation, + health, + }, + ) { + previous.transport.disconnect(); + } + } + + pub(crate) fn accepts(&self, endpoint_id: &ClientEndpointId, generation: u64) -> bool { + self.connections + .get(endpoint_id) + .is_some_and(|connection| connection.generation == generation) + } + + pub(crate) fn received( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + now: Instant, + ) { + if let Some(health) = self + .connections + .get_mut(endpoint_id) + .filter(|connection| connection.generation == generation) + .and_then(|connection| connection.health.as_mut()) + { + health.received(now); + } + } + + pub(crate) fn mark_ready(&mut self, endpoint_id: &ClientEndpointId, generation: u64) { + if let Some(health) = self + .connections + .get_mut(endpoint_id) + .filter(|connection| connection.generation == generation) + .and_then(|connection| connection.health.as_mut()) + { + health.ready(); + } + } + + pub(crate) fn tick_health(&mut self, now: Instant) { + let actions = self + .connections + .iter() + .filter_map(|(endpoint_id, connection)| { + connection + .health + .as_ref() + .map(|health| (endpoint_id.clone(), health.action(now))) + }) + .filter(|(_, action)| *action != HealthAction::None) + .collect::<Vec<_>>(); + for (endpoint_id, action) in actions { + match action { + HealthAction::None => {} + HealthAction::Ping => { + let ping = ClientMessage::EndpointControl { + kind: crate::protocol::endpoint::HEALTH_PING_KIND.into(), + data: String::new(), + }; + if self.send_to(&endpoint_id, &ping) == EndpointSendOutcome::Sent { + if let Some(health) = self + .connections + .get_mut(&endpoint_id) + .and_then(|connection| connection.health.as_mut()) + { + health.ping_sent(now); + } + } + } + HealthAction::Expired => self.record_failure( + endpoint_id, + io::Error::new(io::ErrorKind::TimedOut, "endpoint health check timed out"), + ), + } + } + } + + pub(crate) fn set_active(&mut self, endpoint_id: &ClientEndpointId) -> bool { + if !self + .connections + .get(endpoint_id) + .is_some_and(|connection| connection.surface_active) + { + return false; + } + self.active = endpoint_id.clone(); + true + } + + pub(crate) fn set_surface_active( + &mut self, + endpoint_id: &ClientEndpointId, + active: bool, + ) -> bool { + let Some(connection) = self.connections.get_mut(endpoint_id) else { + return false; + }; + let changed = connection.surface_active != active; + connection.surface_active = active; + changed + } + + pub(crate) fn send(&mut self, message: &ClientMessage) -> EndpointSendOutcome { + let endpoint_id = self.active.clone(); + self.send_to(&endpoint_id, message) + } + + pub(crate) fn send_to( + &mut self, + endpoint_id: &ClientEndpointId, + message: &ClientMessage, + ) -> EndpointSendOutcome { + let result = self + .connections + .get_mut(endpoint_id) + .ok_or_else(|| io::Error::new(io::ErrorKind::NotConnected, "endpoint is unavailable")) + .and_then(|connection| connection.transport.send(message)); + match result { + Ok(()) => EndpointSendOutcome::Sent, + Err(error) => { + self.record_failure(endpoint_id.clone(), error); + EndpointSendOutcome::NotSent + } + } + } + + pub(crate) fn disconnect(&mut self, endpoint_id: &ClientEndpointId) { + self.failures + .retain(|failure| &failure.endpoint_id != endpoint_id); + if let Some(mut connection) = self.connections.remove(endpoint_id) { + connection.transport.disconnect(); + } + } + + pub(crate) fn fail(&mut self, endpoint_id: &ClientEndpointId, error: io::Error) { + self.record_failure(endpoint_id.clone(), error); + } + + pub(crate) fn take_failures(&mut self) -> Vec<EndpointTransportFailure> { + let errors = self + .connections + .iter_mut() + .filter_map(|(id, connection)| { + connection + .transport + .take_error() + .map(|error| (id.clone(), error)) + }) + .collect::<Vec<_>>(); + for (endpoint_id, error) in errors { + self.record_failure(endpoint_id, error); + } + std::mem::take(&mut self.failures) + } + + fn record_failure(&mut self, endpoint_id: ClientEndpointId, error: io::Error) { + let Some(generation) = self + .connections + .get(&endpoint_id) + .map(|connection| connection.generation) + else { + return; + }; + let failure = EndpointTransportFailure { + endpoint_id: endpoint_id.clone(), + generation, + kind: error.kind(), + message: error.to_string(), + }; + if let Some(mut connection) = self.connections.remove(&endpoint_id) { + connection.transport.disconnect(); + } + if let Some(existing) = self + .failures + .iter_mut() + .find(|existing| existing.endpoint_id == endpoint_id) + { + *existing = failure; + } else { + self.failures.push(failure); + } + } +} + +impl Drop for EndpointRegistry { + fn drop(&mut self) { + let deadline = Instant::now() + std::time::Duration::from_millis(250); + for connection in self.connections.values_mut() { + let _ = connection.transport.send(&ClientMessage::Detach); + } + for connection in self.connections.values_mut() { + let _ = connection.transport.flush(deadline); + connection.transport.disconnect(); + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use super::*; + + struct FakeTransport { + sent: Arc<Mutex<Vec<ClientMessage>>>, + error: Option<io::ErrorKind>, + } + + impl EndpointTransport for FakeTransport { + fn send(&mut self, message: &ClientMessage) -> io::Result<()> { + if let Some(kind) = self.error { + return Err(io::Error::new(kind, "fake transport failure")); + } + self.sent.lock().unwrap().push(message.clone()); + Ok(()) + } + } + + fn negotiation() -> EndpointNegotiation { + EndpointNegotiation::new( + vec!["client_shell.surface.set".into()], + vec![ + crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY.into(), + crate::protocol::endpoint::PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(), + crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY.into(), + ], + ) + } + + fn profile() -> crate::client::endpoint::ProfileId { + crate::client::endpoint::ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap() + } + + #[test] + fn endpoint_failures_do_not_remove_other_connections() { + let local_sent = Arc::new(Mutex::new(Vec::new())); + let mut registry = EndpointRegistry::new( + FakeTransport { + sent: local_sent.clone(), + error: None, + }, + 1, + negotiation(), + ); + let ssh_id = ClientEndpointId::Ssh(profile()); + registry.insert( + ssh_id.clone(), + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: Some(io::ErrorKind::BrokenPipe), + }, + 2, + negotiation(), + true, + ); + assert!(registry.set_active(&ssh_id)); + + assert_eq!( + registry.send(&ClientMessage::ClientShellFocus { focused: true }), + EndpointSendOutcome::NotSent + ); + assert!(registry.connection(&ssh_id).is_none()); + assert!(registry.connection(&ClientEndpointId::Local).is_some()); + assert_eq!(registry.take_failures()[0].endpoint_id, ssh_id); + + assert!(registry.set_active(&ClientEndpointId::Local)); + assert_eq!( + registry.send(&ClientMessage::ClientShellFocus { focused: true }), + EndpointSendOutcome::Sent + ); + assert_eq!(local_sent.lock().unwrap().len(), 1); + } + + #[test] + fn reconnecting_active_identity_does_not_count_as_an_active_surface() { + let mut registry = EndpointRegistry::new( + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 1, + negotiation(), + ); + let ssh_id = ClientEndpointId::Ssh(profile()); + registry.insert( + ssh_id.clone(), + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 2, + negotiation(), + true, + ); + assert!(registry.set_active(&ssh_id)); + assert!(registry.active_surface_available()); + registry.disconnect(&ssh_id); + registry.insert( + ssh_id, + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 3, + negotiation(), + false, + ); + assert!(!registry.active_surface_available()); + } + + #[test] + fn recovered_local_uses_transport_failure_not_remote_health_probes() { + let mut registry = EndpointRegistry::empty(); + let sent = Arc::new(Mutex::new(Vec::new())); + registry.insert( + ClientEndpointId::Local, + FakeTransport { + sent: sent.clone(), + error: None, + }, + 2, + negotiation(), + false, + ); + registry.tick_health(Instant::now() + std::time::Duration::from_secs(300)); + assert!(registry.connection(&ClientEndpointId::Local).is_some()); + assert!(sent.lock().unwrap().is_empty()); + assert!(registry.take_failures().is_empty()); + } + + #[test] + fn negotiated_remote_health_probe_expires_the_connection() { + let mut registry = EndpointRegistry::new( + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 1, + negotiation(), + ); + let ssh_id = ClientEndpointId::Ssh(profile()); + let sent = Arc::new(Mutex::new(Vec::new())); + registry.insert( + ssh_id.clone(), + FakeTransport { + sent: sent.clone(), + error: None, + }, + 2, + negotiation(), + false, + ); + let now = Instant::now(); + registry.tick_health(now + super::super::health::HEARTBEAT_INTERVAL); + assert!(matches!( + sent.lock().unwrap().as_slice(), + [ClientMessage::EndpointControl { kind, .. }] + if kind == crate::protocol::endpoint::HEALTH_PING_KIND + )); + + registry.tick_health( + now + super::super::health::HEARTBEAT_INTERVAL + + super::super::health::HEARTBEAT_TIMEOUT, + ); + assert!(registry.connection(&ssh_id).is_none()); + assert_eq!(registry.take_failures()[0].kind, io::ErrorKind::TimedOut); + } + + #[test] + fn a_ready_endpoint_can_stay_connected_after_the_initial_deadline() { + let mut registry = EndpointRegistry::new( + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 1, + negotiation(), + ); + let ssh_id = ClientEndpointId::Ssh(profile()); + registry.insert( + ssh_id.clone(), + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 2, + negotiation(), + false, + ); + let now = Instant::now(); + registry.mark_ready(&ssh_id, 2); + registry.received(&ssh_id, 2, now + super::super::health::HEARTBEAT_INTERVAL); + registry.tick_health(now + super::super::health::HEARTBEAT_TIMEOUT); + assert!(registry.connection(&ssh_id).is_some()); + } + + #[test] + fn negotiated_surface_interest_requires_capability_and_method() { + assert!(negotiation().supports_surface_interest()); + assert!(negotiation().supports_health_check()); + assert!( + !EndpointNegotiation::new(vec!["client_shell.surface.set".into()], Vec::new()) + .supports_surface_interest() + ); + assert!(!EndpointNegotiation::new( + Vec::new(), + vec![crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY.into()] + ) + .supports_surface_interest()); + assert!(!EndpointNegotiation::new( + vec!["client_shell.surface.set".into()], + vec![crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY.into()] + ) + .supports_surface_interest()); + } + + #[test] + fn dropping_registry_detaches_every_connected_endpoint() { + let local_sent = Arc::new(Mutex::new(Vec::new())); + let remote_sent = Arc::new(Mutex::new(Vec::new())); + let mut registry = EndpointRegistry::new( + FakeTransport { + sent: local_sent.clone(), + error: None, + }, + 1, + negotiation(), + ); + registry.insert( + ClientEndpointId::Ssh(profile()), + FakeTransport { + sent: remote_sent.clone(), + error: None, + }, + 2, + negotiation(), + false, + ); + + drop(registry); + + assert!(matches!( + local_sent.lock().unwrap().as_slice(), + [ClientMessage::Detach] + )); + assert!(matches!( + remote_sent.lock().unwrap().as_slice(), + [ClientMessage::Detach] + )); + } + + #[test] + fn stale_generations_are_rejected() { + let registry = EndpointRegistry::new( + FakeTransport { + sent: Arc::new(Mutex::new(Vec::new())), + error: None, + }, + 7, + negotiation(), + ); + assert!(registry.accepts(&ClientEndpointId::Local, 7)); + assert!(!registry.accepts(&ClientEndpointId::Local, 6)); + } +} diff --git a/src/client/endpoint/supervisor.rs b/src/client/endpoint/supervisor.rs new file mode 100644 index 00000000..458ce927 --- /dev/null +++ b/src/client/endpoint/supervisor.rs @@ -0,0 +1,492 @@ +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use super::{ClientEndpointId, ClientEndpointStatus, EndpointNegotiation, NativeEndpointTransport}; +use crate::protocol::{ClientSurfaceSize, RenderEncoding}; +use interprocess::TryClone as _; + +const INITIAL_RETRY_DELAY: Duration = Duration::from_millis(500); +const MAX_RETRY_DELAY: Duration = Duration::from_secs(30); + +#[derive(Clone, Copy)] +pub(crate) struct EndpointConnectOptions { + pub(crate) cols: u16, + pub(crate) rows: u16, + pub(crate) cell_width_px: u32, + pub(crate) cell_height_px: u32, + pub(crate) pixel_geometry_exact: bool, + pub(crate) surface_size: ClientSurfaceSize, + pub(crate) endpoint_keybindings: bool, + pub(crate) mouse_capture: bool, +} + +pub(crate) enum EndpointSupervisorEvent { + Status { + endpoint_id: ClientEndpointId, + generation: u64, + status: ClientEndpointStatus, + message: String, + }, + Connected { + endpoint_id: ClientEndpointId, + generation: u64, + reader: crate::ipc::LocalStream, + writer: NativeEndpointTransport, + negotiation: EndpointNegotiation, + }, +} + +#[derive(Clone)] +enum ConnectTarget { + Local(PathBuf), + Ssh(super::SavedSshEndpoint), +} + +struct ReconnectState { + target: ConnectTarget, + attempts: u32, + next_attempt: Option<Instant>, + in_flight: bool, + generation: Option<u64>, +} + +impl ReconnectState { + fn new(target: ConnectTarget, now: Instant) -> Self { + Self { + target, + attempts: 0, + next_attempt: Some(now), + in_flight: false, + generation: None, + } + } +} + +pub(crate) struct EndpointSupervisors { + endpoints: HashMap<ClientEndpointId, ReconnectState>, + next_generation: u64, + shutdown: Arc<AtomicBool>, +} + +impl EndpointSupervisors { + pub(crate) fn new(profiles: &[super::SavedSshEndpoint], now: Instant) -> Self { + let endpoints = profiles + .iter() + .filter(|profile| profile.enabled) + .map(|profile| { + ( + ClientEndpointId::Ssh(profile.id.clone()), + ReconnectState::new(ConnectTarget::Ssh(profile.clone()), now), + ) + }) + .collect(); + Self { + endpoints, + next_generation: 2, + shutdown: Arc::new(AtomicBool::new(false)), + } + } + + pub(crate) fn add_local(&mut self, path: PathBuf, generation: Option<u64>, now: Instant) { + let mut state = ReconnectState::new(ConnectTarget::Local(path), now); + state.generation = generation; + if generation.is_some() { + state.next_attempt = None; + } + self.endpoints.insert(ClientEndpointId::Local, state); + } + + pub(crate) fn reconcile_profiles( + &mut self, + profiles: &[super::SavedSshEndpoint], + now: Instant, + ) -> Vec<ClientEndpointId> { + let mut retired = Vec::new(); + self.endpoints.retain(|endpoint_id, state| { + let ConnectTarget::Ssh(previous) = &state.target else { + return true; + }; + let keep = profiles.iter().any(|profile| { + profile.id == previous.id + && profile.enabled + && profile.target == previous.target + && profile.session == previous.session + }); + if !keep { + retired.push(endpoint_id.clone()); + } + keep + }); + for profile in profiles.iter().filter(|profile| profile.enabled) { + let state = self + .endpoints + .entry(ClientEndpointId::Ssh(profile.id.clone())) + .or_insert_with(|| ReconnectState::new(ConnectTarget::Ssh(profile.clone()), now)); + state.target = ConnectTarget::Ssh(profile.clone()); + } + retired + } + + pub(crate) fn spawn_due( + &mut self, + now: Instant, + options: EndpointConnectOptions, + event_tx: &tokio::sync::mpsc::Sender<EndpointSupervisorEvent>, + ) { + for (endpoint_id, state) in &mut self.endpoints { + if state.in_flight || state.next_attempt.is_none_or(|deadline| deadline > now) { + continue; + } + state.in_flight = true; + state.next_attempt = None; + let generation = self.next_generation; + state.generation = Some(generation); + self.next_generation = self.next_generation.saturating_add(1); + let endpoint_id = endpoint_id.clone(); + let target = state.target.clone(); + let event_tx = event_tx.clone(); + let shutdown = self.shutdown.clone(); + tokio::spawn(async move { + if shutdown.load(Ordering::Acquire) { + return; + } + let task_endpoint_id = endpoint_id.clone(); + let result = tokio::task::spawn_blocking(move || { + connect_once(&target, options, endpoint_id, generation) + }) + .await; + let event = match result { + Ok(Ok(event)) => event, + Ok(Err(error)) => EndpointSupervisorEvent::Status { + endpoint_id: task_endpoint_id, + generation, + status: if failure_needs_attention(&error) { + ClientEndpointStatus::Attention + } else { + ClientEndpointStatus::Reconnecting + }, + message: error.to_string(), + }, + Err(error) => EndpointSupervisorEvent::Status { + endpoint_id: task_endpoint_id, + generation, + status: ClientEndpointStatus::Reconnecting, + message: format!("endpoint connection task stopped unexpectedly: {error}"), + }, + }; + if !shutdown.load(Ordering::Acquire) { + let _ = event_tx.send(event).await; + } + }); + } + } + + pub(crate) fn record_status( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + status: ClientEndpointStatus, + now: Instant, + ) -> bool { + let Some(state) = self.endpoints.get_mut(endpoint_id) else { + return false; + }; + if state.generation != Some(generation) { + return false; + } + state.in_flight = false; + match status { + ClientEndpointStatus::Online => { + state.attempts = 0; + state.next_attempt = None; + } + ClientEndpointStatus::Attention | ClientEndpointStatus::Disabled => { + state.next_attempt = None + } + ClientEndpointStatus::Connecting | ClientEndpointStatus::Reconnecting => { + state.attempts = state.attempts.saturating_add(1); + state.next_attempt = Some(now + retry_delay(state.attempts)); + } + } + true + } + + pub(crate) fn disconnected( + &mut self, + endpoint_id: &ClientEndpointId, + generation: u64, + now: Instant, + ) -> bool { + self.record_status( + endpoint_id, + generation, + ClientEndpointStatus::Reconnecting, + now, + ) + } +} + +impl Drop for EndpointSupervisors { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::Release); + } +} + +fn connect_once( + target: &ConnectTarget, + options: EndpointConnectOptions, + endpoint_id: ClientEndpointId, + generation: u64, +) -> Result<EndpointSupervisorEvent, std::io::Error> { + let (mut stream, lifetime): (_, Box<dyn Send>) = match target { + ConnectTarget::Local(path) => { + let stream = crate::ipc::connect_local_stream(path).map_err(|error| { + // An absent Local socket is transient, unlike a missing SSH install. + if error.kind() == std::io::ErrorKind::NotFound { + std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "Local is unavailable; start its server to reconnect", + ) + } else { + error + } + })?; + (stream, Box::new(())) + } + ConnectTarget::Ssh(profile) => { + let connected = crate::remote::connect_saved_ssh(profile.id.as_str(), &profile.target, &profile.session).map_err(|error| { + if failure_needs_attention(&error) { + std::io::Error::new(error.kind(), format!("{error}. Run `{}` interactively to approve setup, then restart this client", crate::remote::saved_ssh_bootstrap_command(&profile.target, &profile.session))) + } else { error } + })?; + (connected.stream, Box::new(connected.bridge)) + } + }; + let handshake = super::super::do_handshake( + &mut stream, + options.cols, + options.rows, + options.cell_width_px, + options.cell_height_px, + options.pixel_geometry_exact, + Some(options.surface_size), + options.endpoint_keybindings, + options.mouse_capture, + false, + ) + .map_err(handshake_error)?; + if handshake.encoding != RenderEncoding::SemanticFrame { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "endpoint did not negotiate the semantic client shell", + )); + } + let negotiation = EndpointNegotiation::new( + handshake.endpoint_methods.unwrap_or_default(), + handshake.endpoint_capabilities.unwrap_or_default(), + ); + if !negotiation.supports_surface_interest() + || (!endpoint_id.is_local() && !negotiation.supports_health_check()) + { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "this machine needs a server update before it can participate in multi-machine viewing", + )); + } + let reader = stream.try_clone()?; + let writer = NativeEndpointTransport::with_lifetime(stream, lifetime)?; + Ok(EndpointSupervisorEvent::Connected { + endpoint_id, + generation, + reader, + writer, + negotiation, + }) +} + +fn failure_needs_attention(error: &std::io::Error) -> bool { + crate::remote::saved_ssh_failure_needs_attention(error) +} + +fn handshake_error(error: crate::client::ClientError) -> std::io::Error { + use crate::client::ClientError; + use crate::protocol::FramingError; + match error { + ClientError::ConnectionFailed(error) | ClientError::ConnectionLost(error) => error, + ClientError::HandshakeRejected { error, .. } => { + std::io::Error::new(std::io::ErrorKind::Unsupported, error) + } + ClientError::Protocol(FramingError::Io(error)) => error, + ClientError::Protocol(error) => { + std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string()) + } + ClientError::ServerShutdown { reason } => std::io::Error::new( + std::io::ErrorKind::ConnectionAborted, + reason.unwrap_or_else(|| "server shut down during handshake".into()), + ), + } +} + +fn retry_delay(attempt: u32) -> Duration { + INITIAL_RETRY_DELAY + .saturating_mul( + 1_u32 + .checked_shl(attempt.saturating_sub(1).min(6)) + .unwrap_or(u32::MAX), + ) + .min(MAX_RETRY_DELAY) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::client::endpoint::ProfileId; + + fn profile() -> super::super::SavedSshEndpoint { + super::super::SavedSshEndpoint { + id: ProfileId::parse("0123456789abcdef0123456789abcdef").unwrap(), + label: "Build".into(), + target: "build".into(), + session: "agents".into(), + enabled: true, + } + } + + #[test] + fn live_catalog_preserves_renamed_connections_and_local_recovery() { + let now = Instant::now(); + let mut profile = profile(); + let id = ClientEndpointId::Ssh(profile.id.clone()); + let mut supervisors = EndpointSupervisors::new(&[profile.clone()], now); + supervisors.add_local(PathBuf::from("local"), Some(1), now); + let state = supervisors.endpoints.get_mut(&id).unwrap(); + state.generation = Some(7); + state.attempts = 3; + state.next_attempt = Some(now + Duration::from_secs(4)); + profile.label = "Renamed".into(); + assert!(supervisors.reconcile_profiles(&[profile], now).is_empty()); + let state = &supervisors.endpoints[&id]; + assert_eq!(state.generation, Some(7)); + assert_eq!(state.attempts, 3); + assert_eq!(state.next_attempt, Some(now + Duration::from_secs(4))); + assert_eq!( + supervisors.endpoints[&ClientEndpointId::Local].generation, + Some(1) + ); + } + + #[test] + fn live_catalog_add_disable_enable_and_remove_fence_late_connections() { + let now = Instant::now(); + let mut profile = profile(); + let id = ClientEndpointId::Ssh(profile.id.clone()); + let mut supervisors = EndpointSupervisors::new(&[], now); + assert!(supervisors + .reconcile_profiles(&[profile.clone()], now) + .is_empty()); + assert_eq!(supervisors.endpoints[&id].next_attempt, Some(now)); + supervisors.endpoints.get_mut(&id).unwrap().generation = Some(7); + supervisors.next_generation = 8; + profile.enabled = false; + assert_eq!( + supervisors.reconcile_profiles(&[profile.clone()], now), + vec![id.clone()] + ); + assert!(!supervisors.record_status(&id, 7, ClientEndpointStatus::Online, now)); + profile.enabled = true; + assert!(supervisors + .reconcile_profiles(&[profile.clone()], now) + .is_empty()); + assert!(!supervisors.record_status(&id, 7, ClientEndpointStatus::Online, now)); + assert_eq!(supervisors.next_generation, 8); + assert_eq!(supervisors.reconcile_profiles(&[], now), vec![id.clone()]); + assert!(!supervisors.record_status(&id, 7, ClientEndpointStatus::Online, now)); + } + + #[test] + fn live_catalog_destination_change_retires_only_that_machine() { + let now = Instant::now(); + let mut changed = profile(); + let other = super::super::SavedSshEndpoint::new("Other", "other", "main").unwrap(); + let id = ClientEndpointId::Ssh(changed.id.clone()); + let other_id = ClientEndpointId::Ssh(other.id.clone()); + let mut supervisors = EndpointSupervisors::new(&[changed.clone(), other.clone()], now); + supervisors.endpoints.get_mut(&id).unwrap().generation = Some(2); + supervisors.endpoints.get_mut(&other_id).unwrap().generation = Some(3); + changed.session = "another-session".into(); + assert_eq!( + supervisors.reconcile_profiles(&[changed, other], now), + vec![id.clone()] + ); + assert_eq!(supervisors.endpoints[&id].generation, None); + assert_eq!(supervisors.endpoints[&other_id].generation, Some(3)); + } + + #[test] + fn retry_backoff_is_bounded() { + assert_eq!(retry_delay(1), INITIAL_RETRY_DELAY); + assert_eq!(retry_delay(100), MAX_RETRY_DELAY); + } + + #[test] + fn handshake_network_failures_retry_but_incompatibility_needs_attention() { + let timeout = handshake_error(crate::client::ClientError::ConnectionLost( + std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out"), + )); + assert!(!failure_needs_attention(&timeout)); + let rejected = handshake_error(crate::client::ClientError::HandshakeRejected { + version: 1, + error: "surface capability missing".into(), + }); + assert_eq!(rejected.kind(), std::io::ErrorKind::Unsupported); + assert!(failure_needs_attention(&rejected)); + } + + #[test] + fn healthy_local_only_retries_after_its_connection_fails() { + let now = Instant::now(); + let mut supervisors = EndpointSupervisors::new(&[profile()], now); + supervisors.add_local(PathBuf::from("local.sock"), Some(1), now); + assert!(supervisors.endpoints[&ClientEndpointId::Local] + .next_attempt + .is_none()); + assert!(!supervisors.disconnected(&ClientEndpointId::Local, 0, now)); + assert!(supervisors.endpoints[&ClientEndpointId::Local] + .next_attempt + .is_none()); + assert!(supervisors.disconnected(&ClientEndpointId::Local, 1, now)); + assert_eq!( + supervisors.endpoints[&ClientEndpointId::Local].next_attempt, + Some(now + INITIAL_RETRY_DELAY) + ); + assert_eq!( + supervisors.endpoints[&ClientEndpointId::Ssh(profile().id)].next_attempt, + Some(now) + ); + } + + #[test] + fn ssh_recovery_rejects_stale_generations_and_stops_retries_for_attention() { + let now = Instant::now(); + let mut supervisors = EndpointSupervisors::new(&[profile()], now); + let endpoint_id = ClientEndpointId::Ssh(profile().id); + supervisors + .endpoints + .get_mut(&endpoint_id) + .unwrap() + .generation = Some(4); + assert!(supervisors.record_status(&endpoint_id, 4, ClientEndpointStatus::Online, now)); + assert!(!supervisors.disconnected(&endpoint_id, 3, now)); + assert!(supervisors.endpoints[&endpoint_id].next_attempt.is_none()); + assert!(supervisors.disconnected(&endpoint_id, 4, now)); + assert_eq!( + supervisors.endpoints[&endpoint_id].next_attempt, + Some(now + INITIAL_RETRY_DELAY) + ); + assert!(supervisors.record_status(&endpoint_id, 4, ClientEndpointStatus::Attention, now)); + assert!(supervisors.endpoints[&endpoint_id].next_attempt.is_none()); + } +} diff --git a/src/client/endpoint/writer.rs b/src/client/endpoint/writer.rs new file mode 100644 index 00000000..973ff234 --- /dev/null +++ b/src/client/endpoint/writer.rs @@ -0,0 +1,400 @@ +use std::io; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; +use std::time::{Duration, Instant}; + +use interprocess::local_socket::traits::Stream as _; + +use super::EndpointTransport; +use crate::ipc::LocalStream; +use crate::protocol::ClientMessage; + +const MAX_QUEUED_MESSAGES: usize = 256; +const MAX_QUEUED_BYTES: usize = 2 * crate::protocol::MAX_GRAPHICS_FRAME_SIZE; +const WRITE_TIMEOUT: Duration = Duration::from_secs(5); +const IO_POLL_INTERVAL: Duration = Duration::from_millis(2); + +enum WriterCommand { + Frame(Vec<u8>), + Flush(mpsc::Sender<()>), +} + +/// The UI only enqueues complete frames. A worker owns partial writes, cancellation, and the +/// bridge lifetime, so neither socket backpressure nor bridge teardown can block other endpoints. +pub(crate) struct NativeEndpointTransport { + sender: mpsc::SyncSender<WriterCommand>, + queued_bytes: Arc<AtomicUsize>, + stopped: Arc<AtomicBool>, + error: Arc<Mutex<Option<io::Error>>>, +} + +impl NativeEndpointTransport { + pub(crate) fn with_lifetime( + mut stream: LocalStream, + lifetime: impl Send + 'static, + ) -> io::Result<Self> { + stream.set_nonblocking(true)?; + let (sender, receiver) = mpsc::sync_channel::<WriterCommand>(MAX_QUEUED_MESSAGES); + let queued_bytes = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicBool::new(false)); + let error = Arc::new(Mutex::new(None)); + let worker_bytes = queued_bytes.clone(); + let worker_stop = stopped.clone(); + let worker_error = error.clone(); + std::thread::Builder::new() + .name("endpoint-writer".into()) + .spawn(move || { + let _lifetime = lifetime; + while let Ok(command) = receiver.recv() { + if worker_stop.load(Ordering::Acquire) { + break; + } + let frame = match command { + WriterCommand::Frame(frame) => frame, + WriterCommand::Flush(done) => { + let _ = done.send(()); + continue; + } + }; + let result = write_frame(&mut stream, &frame, &worker_stop); + worker_bytes.fetch_sub(frame.len(), Ordering::AcqRel); + if let Err(error) = result { + if let Ok(mut slot) = worker_error.lock() { + *slot = Some(error); + } + worker_stop.store(true, Ordering::Release); + break; + } + } + })?; + Ok(Self { + sender, + queued_bytes, + stopped, + error, + }) + } + + pub(crate) fn stop_handle(&self) -> Arc<AtomicBool> { + self.stopped.clone() + } +} + +impl EndpointTransport for NativeEndpointTransport { + fn send(&mut self, message: &ClientMessage) -> io::Result<()> { + if self.stopped.load(Ordering::Acquire) { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "endpoint writer stopped", + )); + } + let mut frame = Vec::new(); + crate::protocol::write_message(&mut frame, message) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error.to_string()))?; + let len = frame.len(); + if self + .queued_bytes + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |bytes| { + bytes + .checked_add(len) + .filter(|total| *total <= MAX_QUEUED_BYTES) + }) + .is_err() + { + return Err(queue_full()); + } + self.sender + .try_send(WriterCommand::Frame(frame)) + .map_err(|error| { + self.queued_bytes.fetch_sub(len, Ordering::AcqRel); + match error { + mpsc::TrySendError::Full(_) => queue_full(), + mpsc::TrySendError::Disconnected(_) => { + io::Error::new(io::ErrorKind::BrokenPipe, "endpoint writer stopped") + } + } + }) + } + + fn disconnect(&mut self) { + self.stopped.store(true, Ordering::Release); + } + + fn flush(&mut self, deadline: Instant) -> io::Result<()> { + let (done, completion) = mpsc::channel(); + self.sender + .try_send(WriterCommand::Flush(done)) + .map_err(|_| queue_full())?; + completion + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .map_err(|error| match error { + mpsc::RecvTimeoutError::Timeout => { + io::Error::new(io::ErrorKind::TimedOut, "endpoint flush timed out") + } + mpsc::RecvTimeoutError::Disconnected => { + io::Error::new(io::ErrorKind::BrokenPipe, "endpoint writer stopped") + } + }) + } + + fn take_error(&mut self) -> Option<io::Error> { + self.error.lock().ok()?.take() + } +} + +impl Drop for NativeEndpointTransport { + fn drop(&mut self) { + self.stopped.store(true, Ordering::Release); + } +} + +fn queue_full() -> io::Error { + // A message may already be partially written. Retrying or dropping only this message would + // lose input ordering; revoke the connection and recover through the normal lifecycle. + io::Error::new( + io::ErrorKind::ConnectionAborted, + "endpoint output queue is full", + ) +} + +fn write_frame( + writer: &mut impl io::Write, + mut frame: &[u8], + stopped: &AtomicBool, +) -> io::Result<()> { + let deadline = Instant::now() + WRITE_TIMEOUT; + while !frame.is_empty() && !stopped.load(Ordering::Acquire) { + match writer.write(frame) { + Ok(0) => {} + Ok(written) => { + frame = &frame[written..]; + continue; + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} + Err(error) => return Err(error), + } + if Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "endpoint write timed out", + )); + } + std::thread::sleep(IO_POLL_INTERVAL); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn streams() -> (LocalStream, LocalStream, std::path::PathBuf) { + use interprocess::local_socket::traits::Listener as _; + let path = std::env::temp_dir().join(format!( + "herdr-writer-{}-{}.sock", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + let listener = crate::ipc::bind_private_local_listener(&path).unwrap(); + let accepting = std::thread::spawn(move || listener.accept().unwrap()); + let client = crate::ipc::connect_local_stream(&path).unwrap(); + (client, accepting.join().unwrap(), path) + } + + #[test] + fn native_endpoint_writer_delivers_ordered_protocol_frames() { + let (stream, mut peer, path) = streams(); + let mut transport = NativeEndpointTransport::with_lifetime(stream, ()).unwrap(); + let (done, received) = mpsc::channel(); + let reader = std::thread::spawn(move || { + let first: ClientMessage = + crate::protocol::read_message(&mut peer, crate::protocol::MAX_FRAME_SIZE).unwrap(); + let second: ClientMessage = + crate::protocol::read_message(&mut peer, crate::protocol::MAX_FRAME_SIZE).unwrap(); + done.send((first, second)).unwrap(); + }); + transport + .send(&ClientMessage::ClientShellFocus { focused: true }) + .unwrap(); + transport + .send(&ClientMessage::ClientShellFocus { focused: false }) + .unwrap(); + let messages = received.recv_timeout(Duration::from_secs(3)).unwrap(); + assert_eq!( + messages, + ( + ClientMessage::ClientShellFocus { focused: true }, + ClientMessage::ClientShellFocus { focused: false } + ) + ); + reader.join().unwrap(); + drop(transport); + let _ = std::fs::remove_file(path); + } + + #[test] + fn registry_exit_flushes_queued_input_and_a_complete_detach() { + let (stream, mut peer, path) = streams(); + let transport = NativeEndpointTransport::with_lifetime(stream, ()).unwrap(); + let (done, received) = mpsc::channel(); + let reader = std::thread::spawn(move || { + let result = (|| { + let first: ClientMessage = + crate::protocol::read_message(&mut peer, crate::protocol::MAX_FRAME_SIZE)?; + let second: ClientMessage = + crate::protocol::read_message(&mut peer, crate::protocol::MAX_FRAME_SIZE)?; + Ok::<_, crate::protocol::FramingError>((first, second)) + })(); + done.send(result).unwrap(); + }); + let mut registry = super::super::EndpointRegistry::new( + transport, + 1, + super::super::EndpointNegotiation::default(), + ); + let input = ClientMessage::Input { + data: b"queued input".to_vec(), + }; + assert_eq!( + registry.send(&input), + super::super::EndpointSendOutcome::Sent + ); + drop(registry); + let (first, second) = received + .recv_timeout(Duration::from_secs(3)) + .unwrap() + .expect("clean exit must flush complete frames before closing"); + assert_eq!(first, input); + assert_eq!(second, ClientMessage::Detach); + reader.join().unwrap(); + let _ = std::fs::remove_file(path); + } + + #[test] + fn native_endpoint_flush_drains_large_frames_before_detach() { + let (stream, mut peer, path) = streams(); + let mut transport = NativeEndpointTransport::with_lifetime(stream, ()).unwrap(); + let (done, received) = mpsc::channel(); + let reader = std::thread::spawn(move || { + let first: ClientMessage = + crate::protocol::read_message(&mut peer, crate::protocol::MAX_FRAME_SIZE).unwrap(); + let second: ClientMessage = + crate::protocol::read_message(&mut peer, crate::protocol::MAX_FRAME_SIZE).unwrap(); + done.send((first, second)).unwrap(); + }); + let input = ClientMessage::Input { + data: vec![b'x'; 1024 * 1024], + }; + transport.send(&input).unwrap(); + transport.send(&ClientMessage::Detach).unwrap(); + // Large-frame correctness must not depend on the registry's short exit grace period. + transport + .flush(Instant::now() + Duration::from_secs(10)) + .unwrap(); + drop(transport); + let (first, second) = received.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(first, input); + assert_eq!(second, ClientMessage::Detach); + reader.join().unwrap(); + let _ = std::fs::remove_file(path); + } + + #[test] + fn native_endpoint_teardown_does_not_wait_for_a_stalled_peer() { + struct Lifetime(mpsc::Sender<()>); + impl Drop for Lifetime { + fn drop(&mut self) { + let _ = self.0.send(()); + } + } + let (stream, peer, path) = streams(); + let (done, dropped) = mpsc::channel(); + let mut transport = NativeEndpointTransport::with_lifetime(stream, Lifetime(done)).unwrap(); + transport + .send(&ClientMessage::Input { + data: vec![b'x'; 2 * 1024 * 1024], + }) + .unwrap(); + drop(transport); + dropped + .recv_timeout(Duration::from_secs(3)) + .expect("worker lifetime is released without peer reads"); + drop(peer); + let _ = std::fs::remove_file(path); + } + + #[test] + fn partial_writes_preserve_frame_bytes() { + struct PartialWriter(Vec<u8>); + impl io::Write for PartialWriter { + fn write(&mut self, bytes: &[u8]) -> io::Result<usize> { + let count = bytes.len().min(3); + self.0.extend_from_slice(&bytes[..count]); + Ok(count) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + let mut writer = PartialWriter(Vec::new()); + write_frame(&mut writer, b"first frame", &AtomicBool::new(false)).unwrap(); + write_frame(&mut writer, b"second frame", &AtomicBool::new(false)).unwrap(); + assert_eq!(writer.0, b"first framesecond frame"); + } + + #[test] + fn a_stalled_write_is_cancellable_without_peer_progress() { + struct StalledWriter(mpsc::Sender<()>); + impl io::Write for StalledWriter { + fn write(&mut self, _: &[u8]) -> io::Result<usize> { + let _ = self.0.send(()); + Err(io::ErrorKind::WouldBlock.into()) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + let stop = Arc::new(AtomicBool::new(false)); + let worker_stop = stop.clone(); + let (attempted, attempts) = mpsc::channel(); + let (done, completion) = mpsc::channel(); + let worker = std::thread::spawn(move || { + let result = write_frame(&mut StalledWriter(attempted), b"input", &worker_stop); + done.send(result).unwrap(); + }); + attempts.recv_timeout(Duration::from_secs(2)).unwrap(); + stop.store(true, Ordering::Release); + completion + .recv_timeout(Duration::from_secs(2)) + .unwrap() + .unwrap(); + worker.join().unwrap(); + } + + #[test] + fn a_full_queue_is_a_connection_failure_not_silent_input_loss() { + let (sender, _receiver) = mpsc::sync_channel(1); + let mut transport = NativeEndpointTransport { + sender, + queued_bytes: Arc::new(AtomicUsize::new(0)), + stopped: Arc::new(AtomicBool::new(false)), + error: Arc::new(Mutex::new(None)), + }; + transport.send(&ClientMessage::Detach).unwrap(); + assert_eq!( + transport.send(&ClientMessage::Detach).unwrap_err().kind(), + io::ErrorKind::ConnectionAborted + ); + transport + .queued_bytes + .store(MAX_QUEUED_BYTES, Ordering::Release); + assert_eq!( + transport.send(&ClientMessage::Detach).unwrap_err().kind(), + io::ErrorKind::ConnectionAborted + ); + } +} |