SemianiakaVY/openpencil
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

fix(collab): tolerate public relay handshake latency

Browse source
This commit is contained in:
Kayshen-X 2026-08-02 18:39:21 +08:00
parent 748c6a6c5c
commit 4e423530e9
7 changed files with 404 additions and 130 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
crates/op-collab-transport/src/config.rs
View file

@ -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),

38
crates/op-editor-core/src/collab_panel_hover.rs
View file

@ -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,
})
));
}
}
}

21
crates/op-editor-core/src/collab_ui_state.rs
View file

@ -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::*;

56
crates/op-editor-core/src/collab_ui_state/notice_lifecycle_tests.rs Normal file
View file

@ -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,
})
);
}
}

182
crates/op-host-desktop/src/collab_runtime/network/owner.rs
View file

@ -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<ServerPrelude>,
source: OwnerStreamSource,
}
fn accept_owner_stream(
lan_listener: &TcpListener,
relay: Option<&OwnerRelayRuntime>,
lan_prelude: &Arc<ServerPrelude>,
) -> std::io::Result<Option<(TcpStream, SocketAddr, Arc<ServerPrelude>)>> {
) -> std::io::Result<Option<AcceptedOwnerStream>> {
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<BudgetedFrame> {
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<PeerNetworkCommand>,
) -> 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;

123
crates/op-host-desktop/src/collab_runtime/network/owner_queue_tests.rs Normal file
View file

@ -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<BudgetedFrame> {
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<PeerNetworkCommand>,
) -> 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);
}

108
crates/op-host-desktop/src/collab_runtime/network/owner_relay_timeout_tests.rs Normal file
View file

@ -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();
}