From 6af45d4e3afb392480f88f0a06055210a5129c8a Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Mon, 28 Sep 2026 05:55:56 -0500 Subject: [PATCH 1/3] fix(server): drop the previous session's queued events before every connection discard_stale_session_events ran only before serving a preemption winner. After an ordinary disconnect, EGFX frames, RDPSND waves and clipboard messages the session had queued stayed on the server-global channel, and the next connection wrote them before its own channels existed: a client reconnecting after a drop was sent the old session's EGFX frame on a dynamic channel it had not opened, and dropped. run()'s idle select only discards what it reads before the next accept, and an embedder driving run_connection has no such loop. Discard at the start of every connection, before negotiation, with the same allowlist. --- crates/ironrdp-server/src/server.rs | 64 +++++++++++++++++++++++++---- 1 file changed, 55 insertions(+), 9 deletions(-) diff --git a/crates/ironrdp-server/src/server.rs b/crates/ironrdp-server/src/server.rs index 084c2070a..590585197 100644 --- a/crates/ironrdp-server/src/server.rs +++ b/crates/ironrdp-server/src/server.rs @@ -2054,15 +2054,17 @@ impl RdpServer { /// belongs to the SESSION just replaced, keeping only the small set of /// lifecycle/control events meant to survive across connections. /// - /// Called immediately before serving a preemption winner. The channel is - /// shared across every connection the server ever serves and is read by - /// whichever connection drains it next -- so any event the outgoing - /// session produced but never got around to consuming (its OWN eviction - /// notice, a queued clipboard message, an RDPSND wave, an EGFX frame) - /// would otherwise be delivered to its replacement. That is at best stale - /// (an audio wave from a session that no longer exists) and at worst a - /// real leak (the previous peer's clipboard content, handed unprompted to - /// the client that just replaced it). + /// Called before every connection starts negotiating, and immediately + /// before serving a preemption winner. The channel is shared across every + /// connection the server ever serves and is read by whichever connection + /// drains it next -- so any event the outgoing session produced but never + /// got around to consuming (its OWN eviction notice, a queued clipboard + /// message, an RDPSND wave, an EGFX frame) would otherwise be delivered to + /// its replacement. That is at best stale (an audio wave from a session + /// that no longer exists), at worst a real leak (the previous peer's + /// clipboard content, handed unprompted to the client that just replaced + /// it) or a broken connection (an EGFX frame on a dynamic channel the new + /// client has not opened yet). /// /// An ALLOWLIST of what to KEEP, not a denylist of `EvictedByOtherConnection` /// alone, and deliberately so: this mirrors what `run()`'s own top-level @@ -2262,6 +2264,14 @@ impl RdpServer { // `set_display_suppressed_handle()`. self.display_suppressed.store(false, Ordering::Relaxed); + // A session that ended mid-send leaves what it had queued but not + // consumed on the server-global channel. A client reconnecting within + // milliseconds, before `run()`'s idle loop drains it or with no such + // loop at all when an embedder drives connections itself, was sent + // the previous session's EGFX frame as its first write, on a dynamic + // channel it had not opened yet, and dropped the connection. + self.discard_stale_session_events().await; + let size = self.display.lock().await.size().await; let monitor_count = self.display.lock().await.monitor_count().await; let capabilities = capabilities::capabilities(&self.opts, size); @@ -5594,6 +5604,42 @@ mod preempt_tests { ); } + /// Events the previous session left queued never reach the next + /// connection, including one an embedder drives through `run_connection` + /// without `run()`'s idle loop in between. + #[tokio::test] + async fn a_new_connection_starts_without_the_previous_sessions_events() { + let mut server = RdpServer::builder() + .with_addr((Ipv4Addr::LOCALHOST, 0)) + .with_no_security() + .with_no_input() + .with_no_display() + .build(); + + let sender = server.event_sender().clone(); + let _ = sender.send(ServerEvent::AutoDetectRttRequest); + let _ = sender.send(ServerEvent::Quit("keep me".to_owned())); + + // The client is already gone, so negotiation fails at once; the + // discard happens before it starts. + let (stream, client) = tokio::io::duplex(64); + drop(client); + let _ = server.run_connection(stream).await; + + let remaining = { + let mut rx = server.ev_receiver.lock().await; + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + events + }; + assert!( + matches!(remaining.as_slice(), [ServerEvent::Quit(reason)] if reason == "keep me"), + "only the lifecycle event should be left: {remaining:?}" + ); + } + /// Regression guard for the "no happy-path test" review finding, and the /// load-bearing claim it's actually worried about: that deferring /// `attach_channels` to `serve_negotiated` does NOT silently drop a From 90e31ac5f3bf97a03a9ab0ab75854b6363e97330 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Tue, 29 Sep 2026 15:59:02 -0500 Subject: [PATCH 2/3] review: document the queued-event drop and move its regression test into CI run_connection_with now has a Queued events section saying that on entry it drops everything waiting on the server-global channel except Quit, GetLocalAddr, SetCredentials and SetAutoReconnectCookie, why, and that an embedder should queue events for a connection only after the call has started. run_connection points to it. The regression test lived inline in ironrdp-server, whose [lib] test = false keeps inline tests out of CI. It is now an end-to-end test in ironrdp-testsuite-extra: a Disconnect event is queued before a connection served through run_connection, and a real client must still complete an echo round trip. --- crates/ironrdp-server/src/server.rs | 50 ++------ crates/ironrdp-testsuite-extra/tests/e2e.rs | 128 +++++++++++++++++++- 2 files changed, 138 insertions(+), 40 deletions(-) diff --git a/crates/ironrdp-server/src/server.rs b/crates/ironrdp-server/src/server.rs index 590585197..5ab5d3636 100644 --- a/crates/ironrdp-server/src/server.rs +++ b/crates/ironrdp-server/src/server.rs @@ -2150,7 +2150,8 @@ impl RdpServer { /// connections it accepts itself. /// /// Equivalent to [`run_connection_with`](Self::run_connection_with) with - /// [`TransportTls::Managed`]. + /// [`TransportTls::Managed`], including the treatment of events queued + /// before the call (see `Queued events` there). pub async fn run_connection(&mut self, stream: S) -> ServerResult<()> where S: AsyncRead + AsyncWrite + Send + Sync + Unpin, @@ -2175,6 +2176,17 @@ impl RdpServer { /// Everything past the handshake, including the optional Hybrid CredSSP /// exchange and finalization, is identical to the managed path. /// + /// # Queued events + /// + /// Before negotiation starts, this method drops every event still waiting + /// on the server-global channel, except `Quit`, `GetLocalAddr`, + /// `SetCredentials` and `SetAutoReconnectCookie`. What a previous session + /// queued but never consumed (EGFX frames, RDPSND waves, clipboard + /// messages) would otherwise be written to this client before its + /// channels exist. The same applies to anything an embedder queued through + /// [`event_sender`](Self::event_sender) before calling this method: + /// queue events for a connection only after this method has started. + /// /// # Use case for [`TransportTls::AlreadyDone`] /// /// This mode decouples transport encryption from the RDP security-upgrade @@ -5604,42 +5616,6 @@ mod preempt_tests { ); } - /// Events the previous session left queued never reach the next - /// connection, including one an embedder drives through `run_connection` - /// without `run()`'s idle loop in between. - #[tokio::test] - async fn a_new_connection_starts_without_the_previous_sessions_events() { - let mut server = RdpServer::builder() - .with_addr((Ipv4Addr::LOCALHOST, 0)) - .with_no_security() - .with_no_input() - .with_no_display() - .build(); - - let sender = server.event_sender().clone(); - let _ = sender.send(ServerEvent::AutoDetectRttRequest); - let _ = sender.send(ServerEvent::Quit("keep me".to_owned())); - - // The client is already gone, so negotiation fails at once; the - // discard happens before it starts. - let (stream, client) = tokio::io::duplex(64); - drop(client); - let _ = server.run_connection(stream).await; - - let remaining = { - let mut rx = server.ev_receiver.lock().await; - let mut events = Vec::new(); - while let Ok(event) = rx.try_recv() { - events.push(event); - } - events - }; - assert!( - matches!(remaining.as_slice(), [ServerEvent::Quit(reason)] if reason == "keep me"), - "only the lifecycle event should be left: {remaining:?}" - ); - } - /// Regression guard for the "no happy-path test" review finding, and the /// load-bearing claim it's actually worried about: that deferring /// `attach_channels` to `serve_negotiated` does NOT silently drop a diff --git a/crates/ironrdp-testsuite-extra/tests/e2e.rs b/crates/ironrdp-testsuite-extra/tests/e2e.rs index 7dadba480..8183d25db 100644 --- a/crates/ironrdp-testsuite-extra/tests/e2e.rs +++ b/crates/ironrdp-testsuite-extra/tests/e2e.rs @@ -17,6 +17,7 @@ use ironrdp::pdu::geometry::InclusiveRectangle; use ironrdp::pdu::rdp::capability_sets::MajorPlatformType; use ironrdp::pdu::rdp::client_info::CompressionType as PduCompressionType; use ironrdp::pdu::rdp::headers::CompressionFlags; +use ironrdp::pdu::rdp::server_error_info::{ErrorInfo, ProtocolIndependentCode}; use ironrdp::pdu::{self, gcc}; use ironrdp::server::{ self, Acceptor, DesktopSize, DisplayUpdate, KeyboardEvent, MouseEvent, PixelFormat, RdpServer, RdpServerDisplay, @@ -319,6 +320,71 @@ async fn test_echo_virtual_channel_end_to_end() { .await } +/// A previous session's queued events must not reach the next connection. +/// +/// A `Disconnect` left over from the last session would otherwise end this one +/// as soon as it activates, so the echo round trip never completes. +#[tokio::test] +async fn a_new_connection_starts_without_the_previous_sessions_events() { + let payload = b"ironrdp stale events e2e".to_vec(); + let echo_payload = payload.clone(); + let stale_disconnect = ServerEvent::Disconnect(ErrorInfo::ProtocolIndependentCode( + ProtocolIndependentCode::DisconnectedByOtherconnection, + )); + + client_server_impl( + Some(vec![stale_disconnect]), + default_client_config(), + Vec::new(), + None, + |connector| connector.with_static_channel(DrdynvcClient::new().with_dynamic_channel(EchoClient::new())), + move |mut stage, _activation_factory, mut framed, display_tx, echo_handle| async move { + let _display_tx = display_tx; + let mut image = DecodedImage::new(PixelFormat::RgbA32, DESKTOP_WIDTH, DESKTOP_HEIGHT); + + let deadline = Instant::now() + Duration::from_secs(5); + let mut round_trip_completed = false; + + while !round_trip_completed && Instant::now() < deadline { + echo_handle + .send_request(echo_payload.clone()) + .expect("send echo request"); + + for _ in 0..20 { + if echo_handle + .take_measurements() + .into_iter() + .any(|measurement| measurement.payload == echo_payload) + { + round_trip_completed = true; + break; + } + + let read_result = tokio::time::timeout(Duration::from_millis(150), framed.read_pdu()).await; + let Ok(Ok((action, frame))) = read_result else { + continue; + }; + + let outputs = stage.process(&mut image, action, &frame).expect("stage process"); + for output in outputs { + if let ActiveStageOutput::ResponseFrame(frame) = output { + framed.write_all(&frame).await.expect("write response frame"); + } + } + } + } + + assert!( + round_trip_completed, + "the connection did not survive a Disconnect event queued before it started" + ); + + (stage, framed) + }, + ) + .await +} + #[tokio::test] async fn rdpdr_static_channel_announces_a_drive_and_completes_an_unsupported_create() { let fixture = RdpdrFixtureFactory::new(RdpdrFixtureOperation::UnsupportedCreate); @@ -936,6 +1002,39 @@ async fn client_server_with_connector( + 'static, Fut: Future>>)>, C: FnOnce(connector::ClientConnector) -> connector::ClientConnector + 'static, +{ + client_server_impl( + None, + client_config, + static_channel_factories, + server_udp_addr, + connector_factory, + clientfn, + ) + .await; +} + +/// With `stale_events` set, the server is driven the way an embedder drives +/// it: the events are queued first, then one connection is served through +/// `run_connection`, with no `run()` idle loop in between to discard them. +async fn client_server_impl( + stale_events: Option>, + client_config: connector::Config, + static_channel_factories: Vec>, + server_udp_addr: Option, + connector_factory: C, + clientfn: F, +) where + F: FnOnce( + ActiveStage, + connector::connection_activation::ConnectionActivationFactory, + Framed>>, + UnboundedSender, + server::EchoServerHandle, + ) -> Fut + + 'static, + Fut: Future>>)>, + C: FnOnce(connector::ClientConnector) -> connector::ClientConnector + 'static, { // FIXME(@CBenoit): If this is really necessary, we may consider a non-global way of registering the subscriber; otherwise it’s unnecessary to register that. let _ = tracing_subscriber::fmt() @@ -970,15 +1069,37 @@ async fn client_server_with_connector( let ev = server.event_sender().clone(); let echo_handle = server.echo_handle().clone(); + let direct_listener = match stale_events { + Some(events) => { + for event in events { + ev.send(event).unwrap(); + } + Some(TcpListener::bind(("127.0.0.1", 0)).await.expect("bind listener")) + } + None => None, + }; + let direct_addr = direct_listener + .as_ref() + .map(|listener| listener.local_addr().expect("listener address")); + let local = tokio::task::LocalSet::new(); local .run_until(async move { let server = tokio::task::spawn_local(async move { - server.run().await.unwrap(); + match direct_listener { + Some(listener) => { + let (stream, _) = listener.accept().await.expect("accept connection"); + server.run_connection(stream).await.unwrap(); + } + None => server.run().await.unwrap(), + } }); let client = tokio::task::spawn_local(async move { - let server_addr = local_addr_of(&ev).await; + let server_addr = match direct_addr { + Some(addr) => addr, + None => local_addr_of(&ev).await, + }; let (upgraded_framed, connection_result) = connect_client(server_addr, |client_addr| { connector_factory(connector::ClientConnector::new(client_config, client_addr)) }) @@ -1020,7 +1141,8 @@ async fn client_server_with_connector( while let Ok(pdu) = upgraded_framed.read_pdu().await { debug!(?pdu); } - ev.send(ServerEvent::Quit("bye".into())).unwrap(); + // Nothing is left to receive it when `run_connection` already returned. + let _ = ev.send(ServerEvent::Quit("bye".into())); }); tokio::try_join!(server, client).expect("join"); From 35d1262bfc9d7084ec7168f92e312d65c5b57fb2 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Tue, 29 Sep 2026 17:56:10 -0500 Subject: [PATCH 3/3] review: shorten the drain comment, document the drain limit, share the echo driver The comment at the discard call now says why it runs before negotiation and points to discard_stale_session_events for the rest. The Queued events section says the drain is not a barrier: an event a previous-session producer sends after it has run still reaches the new connection, so producers the embedder runs with its own event_sender clone or one of the server's handles must be stopped first. The two echo round-trip tests share drive_echo_round_trip instead of repeating the client loop. --- crates/ironrdp-server/src/server.rs | 17 ++- crates/ironrdp-testsuite-extra/tests/e2e.rs | 115 ++++++++------------ 2 files changed, 57 insertions(+), 75 deletions(-) diff --git a/crates/ironrdp-server/src/server.rs b/crates/ironrdp-server/src/server.rs index 5ab5d3636..8dc93c4c9 100644 --- a/crates/ironrdp-server/src/server.rs +++ b/crates/ironrdp-server/src/server.rs @@ -2187,6 +2187,14 @@ impl RdpServer { /// [`event_sender`](Self::event_sender) before calling this method: /// queue events for a connection only after this method has started. /// + /// This is a drain, not a barrier: an event that a producer of the previous + /// session sends after it has run still reaches this connection. The + /// backends the server built for that session are dropped before this + /// method returns, but a task the embedder runs with its own + /// [`event_sender`](Self::event_sender) clone or one of the server's + /// handles is not, so stop such producers before starting the next + /// connection. + /// /// # Use case for [`TransportTls::AlreadyDone`] /// /// This mode decouples transport encryption from the RDP security-upgrade @@ -2276,12 +2284,9 @@ impl RdpServer { // `set_display_suppressed_handle()`. self.display_suppressed.store(false, Ordering::Relaxed); - // A session that ended mid-send leaves what it had queued but not - // consumed on the server-global channel. A client reconnecting within - // milliseconds, before `run()`'s idle loop drains it or with no such - // loop at all when an embedder drives connections itself, was sent - // the previous session's EGFX frame as its first write, on a dynamic - // channel it had not opened yet, and dropped the connection. + // Before negotiation, so nothing the previous session left queued can + // reach this client ahead of its own channels. See + // `discard_stale_session_events`. self.discard_stale_session_events().await; let size = self.display.lock().await.size().await; diff --git a/crates/ironrdp-testsuite-extra/tests/e2e.rs b/crates/ironrdp-testsuite-extra/tests/e2e.rs index 8183d25db..679b1ba8f 100644 --- a/crates/ironrdp-testsuite-extra/tests/e2e.rs +++ b/crates/ironrdp-testsuite-extra/tests/e2e.rs @@ -264,6 +264,46 @@ fn compressed_bitmap_fastpath_frame(compressor: &mut BulkCompressor) -> (Vec (frame, flags) } +/// Sends echo requests and serves the client session until the server reports +/// the round trip for `payload`, or five seconds pass. +async fn drive_echo_round_trip( + stage: &mut ActiveStage, + framed: &mut Framed>>, + echo_handle: &server::EchoServerHandle, + payload: &[u8], +) -> Option { + let mut image = DecodedImage::new(PixelFormat::RgbA32, DESKTOP_WIDTH, DESKTOP_HEIGHT); + let deadline = Instant::now() + Duration::from_secs(5); + + while Instant::now() < deadline { + echo_handle.send_request(payload.to_vec()).expect("send echo request"); + + for _ in 0..20 { + if let Some(measurement) = echo_handle + .take_measurements() + .into_iter() + .find(|measurement| measurement.payload == payload) + { + return Some(measurement); + } + + let read_result = tokio::time::timeout(Duration::from_millis(150), framed.read_pdu()).await; + let Ok(Ok((action, frame))) = read_result else { + continue; + }; + + let outputs = stage.process(&mut image, action, &frame).expect("stage process"); + for output in outputs { + if let ActiveStageOutput::ResponseFrame(frame) = output { + framed.write_all(&frame).await.expect("write response frame"); + } + } + } + } + + None +} + #[tokio::test] async fn test_echo_virtual_channel_end_to_end() { let payload = b"ironrdp echo e2e".to_vec(); @@ -276,42 +316,10 @@ async fn test_echo_virtual_channel_end_to_end() { |connector| connector.with_static_channel(DrdynvcClient::new().with_dynamic_channel(EchoClient::new())), move |mut stage, _activation_factory, mut framed, display_tx, echo_handle| async move { let _display_tx = display_tx; - let mut image = DecodedImage::new(PixelFormat::RgbA32, DESKTOP_WIDTH, DESKTOP_HEIGHT); - - let deadline = Instant::now() + Duration::from_secs(5); - let mut matched_measurement = None; - - while Instant::now() < deadline { - echo_handle - .send_request(echo_payload.clone()) - .expect("send echo request"); - - for _ in 0..20 { - let measurements = echo_handle.take_measurements(); - if let Some(measurement) = measurements.into_iter().find(|m| m.payload == echo_payload) { - matched_measurement = Some(measurement); - break; - } - - let read_result = tokio::time::timeout(Duration::from_millis(150), framed.read_pdu()).await; - let Ok(Ok((action, frame))) = read_result else { - continue; - }; - let outputs = stage.process(&mut image, action, &frame).expect("stage process"); - for output in outputs { - if let ActiveStageOutput::ResponseFrame(frame) = output { - framed.write_all(&frame).await.expect("write response frame"); - } - } - } - - if matched_measurement.is_some() { - break; - } - } - - let measurement = matched_measurement.expect("echo RTT measurement was not produced"); + let measurement = drive_echo_round_trip(&mut stage, &mut framed, &echo_handle, &echo_payload) + .await + .expect("echo RTT measurement was not produced"); assert_eq!(measurement.payload, echo_payload); (stage, framed) @@ -340,42 +348,11 @@ async fn a_new_connection_starts_without_the_previous_sessions_events() { |connector| connector.with_static_channel(DrdynvcClient::new().with_dynamic_channel(EchoClient::new())), move |mut stage, _activation_factory, mut framed, display_tx, echo_handle| async move { let _display_tx = display_tx; - let mut image = DecodedImage::new(PixelFormat::RgbA32, DESKTOP_WIDTH, DESKTOP_HEIGHT); - - let deadline = Instant::now() + Duration::from_secs(5); - let mut round_trip_completed = false; - - while !round_trip_completed && Instant::now() < deadline { - echo_handle - .send_request(echo_payload.clone()) - .expect("send echo request"); - - for _ in 0..20 { - if echo_handle - .take_measurements() - .into_iter() - .any(|measurement| measurement.payload == echo_payload) - { - round_trip_completed = true; - break; - } - - let read_result = tokio::time::timeout(Duration::from_millis(150), framed.read_pdu()).await; - let Ok(Ok((action, frame))) = read_result else { - continue; - }; - - let outputs = stage.process(&mut image, action, &frame).expect("stage process"); - for output in outputs { - if let ActiveStageOutput::ResponseFrame(frame) = output { - framed.write_all(&frame).await.expect("write response frame"); - } - } - } - } assert!( - round_trip_completed, + drive_echo_round_trip(&mut stage, &mut framed, &echo_handle, &echo_payload) + .await + .is_some(), "the connection did not survive a Disconnect event queued before it started" );