diff --git a/crates/op-collab-transport/src/config.rs b/crates/op-collab-transport/src/config.rs index 5d8e61afe..1d1e00608 100644 --- a/crates/op-collab-transport/src/config.rs +++ b/crates/op-collab-transport/src/config.rs @@ -104,9 +104,9 @@ impl Default for TimeoutConfig { connect: Duration::from_secs(5), handshake: Duration::from_secs(5), // One round trip after TCP establishment is enough for an honest - // initiator, including relay-mediated paths, while keeping a silent - // peer's hold on a global pending seat to a fifth of the handshake - // window. + // directly accepted initiator, while keeping a silent peer's hold + // on a global pending seat to a fifth of the handshake window. + // Authenticated relay bridges may widen this per accepted stream. handshake_first_message: Duration::from_secs(1), admission: Duration::from_secs(10), ordinary_transfer: Duration::from_secs(10), diff --git a/crates/op-editor-core/src/collab_panel_hover.rs b/crates/op-editor-core/src/collab_panel_hover.rs index f455f98d6..24967dd05 100644 --- a/crates/op-editor-core/src/collab_panel_hover.rs +++ b/crates/op-editor-core/src/collab_panel_hover.rs @@ -41,6 +41,12 @@ impl crate::CollabUiState { /// Update phase and invalidate feedback whose action may have moved under /// a stationary cursor. pub fn set_phase(&mut self, phase: crate::CollabConnectionPhase) { + if matches!( + phase, + crate::CollabConnectionPhase::Starting | crate::CollabConnectionPhase::Joining + ) { + self.clear_connect_notice(); + } if self.phase != phase { self.panel.hover = None; self.phase = phase; @@ -51,7 +57,10 @@ impl crate::CollabUiState { #[cfg(test)] mod tests { use super::CollabPanelHover; - use crate::{CollabAvailability, CollabNoticeKind, CollabUiState}; + use crate::{ + CollabAvailability, CollabConnectErrorUi, CollabConnectionPhase, CollabNoticeKind, + CollabRejectUiCode, CollabUiState, + }; #[test] fn screen_and_notice_changes_clear_stale_panel_hover() { @@ -64,4 +73,31 @@ mod tests { state.set_availability(CollabAvailability::SignInRequired); assert_eq!(state.panel.hover, None); } + + #[test] + fn a_new_connection_attempt_retires_only_connect_notices() { + for phase in [ + CollabConnectionPhase::Starting, + CollabConnectionPhase::Joining, + ] { + let mut stale_connect = CollabUiState::default(); + stale_connect.set_notice( + CollabNoticeKind::Connect(CollabConnectErrorUi::RelayUnavailable), + 7, + ); + stale_connect.set_phase(phase); + assert_eq!(stale_connect.notice, None); + + let mut session_notice = CollabUiState::default(); + session_notice.set_notice(CollabNoticeKind::Reject(CollabRejectUiCode::Conflict), 9); + session_notice.set_phase(phase); + assert!(matches!( + session_notice.notice, + Some(crate::CollabNotice { + kind: CollabNoticeKind::Reject(CollabRejectUiCode::Conflict), + created_at_ms: 9, + }) + )); + } + } } diff --git a/crates/op-editor-core/src/collab_ui_state.rs b/crates/op-editor-core/src/collab_ui_state.rs index 9684a252e..149638174 100644 --- a/crates/op-editor-core/src/collab_ui_state.rs +++ b/crates/op-editor-core/src/collab_ui_state.rs @@ -431,6 +431,14 @@ impl CollabUiState { session.share_endpoint = None; } self.set_phase(phase); + if phase == CollabConnectionPhase::Active { + // A connect notice describes the attempt that just completed. It + // must not survive into the authenticated session, while + // session-scoped notices (rejects, conflicts, owner departure, + // and so on) remain actionable until their own lifecycle retires + // them. + self.clear_connect_notice(); + } // The join decision has been made; retiring the projection keeps a // stale prompt from reappearing over a live session. self.clear_owner_confirmation(); @@ -582,6 +590,16 @@ impl CollabUiState { }); } + pub(crate) fn clear_connect_notice(&mut self) { + if self + .notice + .is_some_and(|notice| matches!(notice.kind, CollabNoticeKind::Connect(_))) + { + self.panel.hover = None; + self.notice = None; + } + } + fn retain_rostered_presence(&mut self) { let participants = &self.participants; let mut presence = self.presence.as_ref().clone(); @@ -604,6 +622,9 @@ impl CollabUiState { #[cfg(test)] mod presence_merge_tests; +#[cfg(test)] +mod notice_lifecycle_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/crates/op-editor-core/src/collab_ui_state/notice_lifecycle_tests.rs b/crates/op-editor-core/src/collab_ui_state/notice_lifecycle_tests.rs new file mode 100644 index 000000000..90f6febd7 --- /dev/null +++ b/crates/op-editor-core/src/collab_ui_state/notice_lifecycle_tests.rs @@ -0,0 +1,56 @@ +use super::*; +use crate::{CollabConnectErrorUi, CollabRejectUiCode}; + +fn authenticated_session() -> AuthenticatedCollabSession { + AuthenticatedCollabSession { + session_name: "Shared design".to_string(), + role: CollabUiRole::Editor, + share_endpoint: None, + } +} + +#[test] +fn authenticated_active_retires_stale_connect_notice() { + let mut state = CollabUiState::default(); + state.set_notice( + CollabNoticeKind::Connect(CollabConnectErrorUi::RelayUnavailable), + 7, + ); + + assert!(state.set_authenticated_session( + CollabConnectionPhase::Active, + authenticated_session(), + Vec::new(), + )); + + assert_eq!(state.notice, None); +} + +#[test] +fn authenticated_active_preserves_session_notices() { + let notices = [ + CollabNoticeKind::Reject(CollabRejectUiCode::Conflict), + CollabNoticeKind::EditConflictDiscarded, + CollabNoticeKind::OwnerLeft, + CollabNoticeKind::DisconnectedReadOnly, + ]; + + for (index, kind) in notices.into_iter().enumerate() { + let mut state = CollabUiState::default(); + state.set_notice(kind, index as u64); + + assert!(state.set_authenticated_session( + CollabConnectionPhase::Active, + authenticated_session(), + Vec::new(), + )); + + assert_eq!( + state.notice, + Some(CollabNotice { + kind, + created_at_ms: index as u64, + }) + ); + } +} diff --git a/crates/op-host-desktop/src/collab_runtime/network/owner.rs b/crates/op-host-desktop/src/collab_runtime/network/owner.rs index 7fd927ff7..043212218 100644 --- a/crates/op-host-desktop/src/collab_runtime/network/owner.rs +++ b/crates/op-host-desktop/src/collab_runtime/network/owner.rs @@ -217,7 +217,13 @@ fn run_inner( } } match accept_owner_stream(&listener, relay.as_ref(), &prelude) { - Ok(Some((stream, address, accepted_prelude))) => { + Ok(Some(accepted)) => { + let AcceptedOwnerStream { + stream, + address, + prelude: accepted_prelude, + source, + } = accepted; let Some(raw_connection) = next_connection else { continue; }; @@ -242,7 +248,7 @@ fn run_inner( verifier: Arc::clone(&verifier), prelude: accepted_prelude, shared_budget: shared_budget.clone(), - config, + config: source.transport_config(config), connection_id, session_id: session_id.clone(), epoch, @@ -283,22 +289,58 @@ fn run_inner( } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum OwnerStreamSource { + Lan, + Relay, +} + +impl OwnerStreamSource { + fn transport_config(self, mut config: TransportConfig) -> TransportConfig { + if self == Self::Relay { + // The relay already authenticated and bounded the public socket. + // Its local bridge may receive the initiator's first Noise frame + // after a public-network round trip, so retain the complete Noise + // handshake window instead of applying the direct-LAN DoS guard. + config.timeouts.handshake_first_message = config.timeouts.handshake; + } + config + } +} + +struct AcceptedOwnerStream { + stream: TcpStream, + address: SocketAddr, + prelude: Arc, + source: OwnerStreamSource, +} + fn accept_owner_stream( lan_listener: &TcpListener, relay: Option<&OwnerRelayRuntime>, lan_prelude: &Arc, -) -> std::io::Result)>> { +) -> std::io::Result> { if let Some(relay) = relay { match relay.accept() { Ok((stream, address)) => { - return Ok(Some((stream, address, relay.prelude()))); + return Ok(Some(AcceptedOwnerStream { + stream, + address, + prelude: relay.prelude(), + source: OwnerStreamSource::Relay, + })); } Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {} Err(error) => return Err(error), } } match lan_listener.accept() { - Ok((stream, address)) => Ok(Some((stream, address, Arc::clone(lan_prelude)))), + Ok((stream, address)) => Ok(Some(AcceptedOwnerStream { + stream, + address, + prelude: Arc::clone(lan_prelude), + source: OwnerStreamSource::Lan, + })), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => Ok(None), Err(error) => Err(error), } @@ -577,45 +619,9 @@ fn terminal_rotation_failure(error: &DiscoveryError, publisher_stopped: bool) -> #[cfg(test)] mod approval_timeout_tests { - use op_collab::{Bye, ByeReason, CollabMessage, Epoch, FrameEnvelope, Presence, SessionId}; - use op_collab_transport::{encode_frame_transfer, m1_wire_limits, EncodedFrameTransfer}; + use op_collab::ByeReason; use super::*; - use crate::collab_runtime::types::BudgetedFrame; - - fn frame(message: CollabMessage) -> FrameEnvelope { - FrameEnvelope::new(SessionId::from("bridge-routing"), Epoch(1), message) - } - - fn budgeted(frame: FrameEnvelope, budget: &SharedQueueBudget) -> Box { - let lossy = crate::collab_runtime::types::is_lossy_presence_frame(&frame); - let encoded = EncodedFrameTransfer::encode(&frame, m1_wire_limits()).unwrap(); - let encoded_len = encoded.encoded_len(); - Box::new(BudgetedFrame::new( - encoded, - lossy, - budget.reserve(encoded_len).unwrap(), - )) - } - - fn active_peer_registry( - connection: ConnectionKey, - commands: std::sync::mpsc::SyncSender, - ) -> PeerRegistry { - let (shutdown, _shutdown_receiver) = mpsc::sync_channel(1); - let mut peers = PeerRegistry::new(); - peers.insert( - connection, - PeerControl { - commands, - shutdown, - cancel: None, - phase: Arc::new(AtomicU8::new(PeerPhase::Active as u8)), - thread: None, - }, - ); - peers - } #[test] fn owner_approval_window_is_human_sized_but_bounded() { @@ -700,88 +706,12 @@ mod approval_timeout_tests { TcpStream::connect(endpoint).expect("IPv4 loopback reaches fallback policy"); listener.accept().expect("accept IPv4 loopback"); } - - #[test] - fn owner_outer_and_peer_handoffs_retain_one_shared_reservation() { - let reliable = frame(CollabMessage::Bye(Bye { - reason: ByeReason::Normal, - })); - let encoded_len = encode_frame_transfer(&reliable, m1_wire_limits()) - .unwrap() - .1 - .len(); - let budget = SharedQueueBudget::new(encoded_len).unwrap(); - let connection = ConnectionKey::new(2).unwrap(); - let (outer_sender, outer_receiver) = mpsc::sync_channel(1); - let (peer_sender, peer_receiver) = mpsc::sync_channel(1); - let mut peers = active_peer_registry(connection, peer_sender); - - outer_sender - .send(OwnerNetworkCommand::Send { - connection, - frame: budgeted(reliable, &budget), - coalesce_key: None, - }) - .unwrap(); - assert_eq!(budget.used().unwrap(), encoded_len); - assert!(budget.reserve(1).is_err()); - - let command = outer_receiver.recv().unwrap(); - assert!(route_command(command, &mut peers).unwrap()); - assert_eq!(budget.used().unwrap(), encoded_len); - assert!(budget.reserve(1).is_err()); - - let peer_command = peer_receiver.recv().unwrap(); - assert_eq!(budget.used().unwrap(), encoded_len); - drop(peer_command); - assert_eq!(budget.used().unwrap(), 0); - } - - #[test] - fn full_peer_lane_fails_reliable_but_drops_presence() { - let reliable = frame(CollabMessage::Bye(Bye { - reason: ByeReason::Normal, - })); - let presence = frame(CollabMessage::PresenceUpdate(Presence { - cursor: None, - selection: Vec::new(), - viewport: None, - editing_node: None, - })); - let reliable_len = encode_frame_transfer(&reliable, m1_wire_limits()) - .unwrap() - .1 - .len(); - let presence_len = encode_frame_transfer(&presence, m1_wire_limits()) - .unwrap() - .1 - .len(); - let budget = SharedQueueBudget::new(reliable_len.max(presence_len)).unwrap(); - let connection = ConnectionKey::new(3).unwrap(); - let (peer_sender, _peer_receiver) = mpsc::sync_channel(1); - peer_sender.try_send(PeerNetworkCommand::Stop).unwrap(); - let mut peers = active_peer_registry(connection, peer_sender); - - let reliable_result = route_command( - OwnerNetworkCommand::Send { - connection, - frame: budgeted(reliable, &budget), - coalesce_key: None, - }, - &mut peers, - ); - assert_eq!(reliable_result, Err(CollabRuntimeFailure::ResourceLimit)); - assert_eq!(budget.used().unwrap(), 0); - - assert!(route_command( - OwnerNetworkCommand::Send { - connection, - frame: budgeted(presence, &budget), - coalesce_key: Some(1), - }, - &mut peers, - ) - .unwrap()); - assert_eq!(budget.used().unwrap(), 0); - } } + +#[cfg(test)] +#[path = "owner_queue_tests.rs"] +mod queue_tests; + +#[cfg(test)] +#[path = "owner_relay_timeout_tests.rs"] +mod relay_timeout_tests; diff --git a/crates/op-host-desktop/src/collab_runtime/network/owner_queue_tests.rs b/crates/op-host-desktop/src/collab_runtime/network/owner_queue_tests.rs new file mode 100644 index 000000000..3674ab768 --- /dev/null +++ b/crates/op-host-desktop/src/collab_runtime/network/owner_queue_tests.rs @@ -0,0 +1,123 @@ +use op_collab::{Bye, ByeReason, CollabMessage, Epoch, FrameEnvelope, Presence, SessionId}; +use op_collab_transport::{encode_frame_transfer, m1_wire_limits, EncodedFrameTransfer}; + +use super::*; +use crate::collab_runtime::types::BudgetedFrame; + +fn frame(message: CollabMessage) -> FrameEnvelope { + FrameEnvelope::new(SessionId::from("bridge-routing"), Epoch(1), message) +} + +fn budgeted(frame: FrameEnvelope, budget: &SharedQueueBudget) -> Box { + let lossy = crate::collab_runtime::types::is_lossy_presence_frame(&frame); + let encoded = EncodedFrameTransfer::encode(&frame, m1_wire_limits()).unwrap(); + let encoded_len = encoded.encoded_len(); + Box::new(BudgetedFrame::new( + encoded, + lossy, + budget.reserve(encoded_len).unwrap(), + )) +} + +fn active_peer_registry( + connection: ConnectionKey, + commands: std::sync::mpsc::SyncSender, +) -> PeerRegistry { + let (shutdown, _shutdown_receiver) = mpsc::sync_channel(1); + let mut peers = PeerRegistry::new(); + peers.insert( + connection, + PeerControl { + commands, + shutdown, + cancel: None, + phase: Arc::new(AtomicU8::new(PeerPhase::Active as u8)), + thread: None, + }, + ); + peers +} + +#[test] +fn owner_outer_and_peer_handoffs_retain_one_shared_reservation() { + let reliable = frame(CollabMessage::Bye(Bye { + reason: ByeReason::Normal, + })); + let encoded_len = encode_frame_transfer(&reliable, m1_wire_limits()) + .unwrap() + .1 + .len(); + let budget = SharedQueueBudget::new(encoded_len).unwrap(); + let connection = ConnectionKey::new(2).unwrap(); + let (outer_sender, outer_receiver) = mpsc::sync_channel(1); + let (peer_sender, peer_receiver) = mpsc::sync_channel(1); + let mut peers = active_peer_registry(connection, peer_sender); + + outer_sender + .send(OwnerNetworkCommand::Send { + connection, + frame: budgeted(reliable, &budget), + coalesce_key: None, + }) + .unwrap(); + assert_eq!(budget.used().unwrap(), encoded_len); + assert!(budget.reserve(1).is_err()); + + let command = outer_receiver.recv().unwrap(); + assert!(route_command(command, &mut peers).unwrap()); + assert_eq!(budget.used().unwrap(), encoded_len); + assert!(budget.reserve(1).is_err()); + + let peer_command = peer_receiver.recv().unwrap(); + assert_eq!(budget.used().unwrap(), encoded_len); + drop(peer_command); + assert_eq!(budget.used().unwrap(), 0); +} + +#[test] +fn full_peer_lane_fails_reliable_but_drops_presence() { + let reliable = frame(CollabMessage::Bye(Bye { + reason: ByeReason::Normal, + })); + let presence = frame(CollabMessage::PresenceUpdate(Presence { + cursor: None, + selection: Vec::new(), + viewport: None, + editing_node: None, + })); + let reliable_len = encode_frame_transfer(&reliable, m1_wire_limits()) + .unwrap() + .1 + .len(); + let presence_len = encode_frame_transfer(&presence, m1_wire_limits()) + .unwrap() + .1 + .len(); + let budget = SharedQueueBudget::new(reliable_len.max(presence_len)).unwrap(); + let connection = ConnectionKey::new(3).unwrap(); + let (peer_sender, _peer_receiver) = mpsc::sync_channel(1); + peer_sender.try_send(PeerNetworkCommand::Stop).unwrap(); + let mut peers = active_peer_registry(connection, peer_sender); + + let reliable_result = route_command( + OwnerNetworkCommand::Send { + connection, + frame: budgeted(reliable, &budget), + coalesce_key: None, + }, + &mut peers, + ); + assert_eq!(reliable_result, Err(CollabRuntimeFailure::ResourceLimit)); + assert_eq!(budget.used().unwrap(), 0); + + assert!(route_command( + OwnerNetworkCommand::Send { + connection, + frame: budgeted(presence, &budget), + coalesce_key: Some(1), + }, + &mut peers, + ) + .unwrap()); + assert_eq!(budget.used().unwrap(), 0); +} diff --git a/crates/op-host-desktop/src/collab_runtime/network/owner_relay_timeout_tests.rs b/crates/op-host-desktop/src/collab_runtime/network/owner_relay_timeout_tests.rs new file mode 100644 index 000000000..860e96f41 --- /dev/null +++ b/crates/op-host-desktop/src/collab_runtime/network/owner_relay_timeout_tests.rs @@ -0,0 +1,108 @@ +use std::io::{Read, Write}; +use std::net::Shutdown; + +use op_collab::{Epoch, SessionId}; +use op_collab_transport::{connect_secure_tcp, ConnectionLimiter, TimeoutConfig}; + +use super::*; + +#[test] +fn relay_stream_uses_the_complete_handshake_window_for_its_first_noise_frame() { + let mut config = TransportConfig::default(); + config.timeouts.handshake = Duration::from_secs(7); + config.timeouts.handshake_first_message = Duration::from_millis(250); + + let relay = OwnerStreamSource::Relay.transport_config(config); + + assert_eq!( + relay.timeouts.handshake_first_message, + relay.timeouts.handshake + ); + assert_eq!(relay.timeouts.handshake, Duration::from_secs(7)); + assert_eq!(relay.connections, config.connections); + assert_eq!(relay.rate, config.rate); + assert_eq!(relay.wire_limits, config.wire_limits); +} + +#[test] +fn lan_stream_keeps_the_short_first_noise_frame_guard() { + let config = TransportConfig::default(); + + assert_eq!(OwnerStreamSource::Lan.transport_config(config), config); + assert!(config.timeouts.handshake_first_message < config.timeouts.handshake); +} + +#[test] +fn relay_stream_accepts_a_first_noise_frame_after_the_lan_guard_expires() { + let lan_first_message = Duration::from_millis(50); + let relay_delay = Duration::from_millis(200); + let config = TransportConfig { + timeouts: TimeoutConfig { + handshake: Duration::from_secs(2), + handshake_first_message: lan_first_message, + ..TimeoutConfig::default() + }, + ..TransportConfig::default() + }; + let relay_config = OwnerStreamSource::Relay.transport_config(config); + let prelude = ServerPrelude::new( + "00112233445566778899aabbccddeeff".to_owned(), + SessionId::from("relay-delayed-noise"), + Epoch(1), + ) + .unwrap(); + let expected_discovery_id = prelude.discovery_id().to_owned(); + + let owner_listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let owner_address = owner_listener.local_addr().unwrap(); + let proxy_listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let proxy_address = proxy_listener.local_addr().unwrap(); + let limiter = ConnectionLimiter::with_timeouts(config.connections, config.timeouts).unwrap(); + let limiter_for_owner = limiter.clone(); + let owner = std::thread::spawn(move || { + let (stream, peer) = owner_listener.accept().unwrap(); + let pending = limiter_for_owner.try_begin_handshake(peer.ip()).unwrap(); + accept_secure_tcp_guarded( + stream, + &DeviceStaticKey::from_private([41_u8; 32]).unwrap(), + &prelude, + relay_config, + &pending, + ) + .is_ok() + }); + let proxy = std::thread::spawn(move || { + let (mut guest_stream, _) = proxy_listener.accept().unwrap(); + let mut owner_stream = TcpStream::connect(owner_address).unwrap(); + let mut owner_reader = owner_stream.try_clone().unwrap(); + let mut guest_writer = guest_stream.try_clone().unwrap(); + let downstream = std::thread::spawn(move || { + let _ = std::io::copy(&mut owner_reader, &mut guest_writer); + let _ = guest_writer.shutdown(Shutdown::Write); + }); + + let mut first_frame = [0_u8; 8 * 1024]; + let first_len = guest_stream.read(&mut first_frame).unwrap(); + assert!(first_len > 0); + std::thread::sleep(relay_delay); + owner_stream.write_all(&first_frame[..first_len]).unwrap(); + let _ = std::io::copy(&mut guest_stream, &mut owner_stream); + let _ = owner_stream.shutdown(Shutdown::Write); + downstream.join().unwrap(); + }); + + let started = Instant::now(); + let (_, connection) = connect_secure_tcp( + proxy_address, + &DeviceStaticKey::from_private([42_u8; 32]).unwrap(), + Some(&expected_discovery_id), + config, + ) + .unwrap(); + assert!(started.elapsed() >= relay_delay); + assert!(started.elapsed() > lan_first_message); + drop(connection); + + assert!(owner.join().unwrap()); + proxy.join().unwrap(); +}