ConnectionPolicy::Preempt drops ConnectionHandler hooks fired during the served connection (on_connection_info never fires)

[clintcan]

Summary

Under ConnectionPolicy::Preempt, ConnectionHandler::on_connection_info never
reaches the installed handler — for every served connection, not only ones
that end up preempted. on_accept still does. Queue and Reject are
unaffected, and Queue is the default, so today this only hits embedders who opt
into Preempt — but for them the hook is silently dead, with nothing logged.

That breaks the trait's documented contract, which says on_connection_info "is
called from every code path that completes connection setup" and is "called once
per connection".

Reproduction

Three e2e tests that drive a real client through the full handshake via
RdpServer::run(), identical except for the policy:

test e2e::connection_handler_hooks_reach_handler_under_queue ... ok
test e2e::connection_handler_hooks_reach_handler_under_reject ... ok
test e2e::connection_handler_hooks_reach_handler_under_preempt ... FAILED
Preempt: on_accept=true, on_connection_info=false

The client completes connect_finalize in all three, and under Preempt the
same handler still receives on_accept — it just never receives
on_connection_info. Deterministic across repeated runs.

To confirm the cause rather than just correlate it, I temporarily stopped the
Preempt arm from taking the handler (see below): all three then pass, and the
Preempt test finishes immediately instead of waiting out its deadline.

Tests — appended to crates/ironrdp-testsuite-extra/tests/e2e.rs on d2bb7376 (clippy- and fmt-clean)

/// Records which `ConnectionHandler` hooks reached the handler. `on_accept` fires
/// before a connection is served; `on_connection_info` fires mid-connection.
struct HookRecorder {
    accept: Arc<AtomicBool>,
    info: Arc<AtomicBool>,
}

impl server::ConnectionHandler for HookRecorder {
    fn on_accept(&mut self, _peer: core::net::SocketAddr) -> bool {
        self.accept.store(true, Ordering::SeqCst);
        true
    }

    fn on_connection_info(&mut self, _info: &server::ConnectionInfo) {
        self.info.store(true, Ordering::SeqCst);
    }
}

/// Serve one real client through the full handshake under `policy` and report
/// `(on_accept reached the handler, on_connection_info reached the handler)`.
async fn hooks_reaching_handler_under(policy: server::ConnectionPolicy) -> (bool, bool) {
    let cert_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/certs/server-cert.pem");
    let key_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/certs/server-key.pem");
    let identity = TlsIdentityCtx::init_from_paths(&cert_path, &key_path).expect("failed to init TLS identity");
    let acceptor = identity.make_acceptor().expect("failed to build TLS acceptor");

    let accept = Arc::new(AtomicBool::new(false));
    let fired = Arc::new(AtomicBool::new(false));
    let (_display_tx, display_rx) = mpsc::unbounded_channel();
    let mut server = RdpServer::builder()
        .with_addr(([127, 0, 0, 1], 0))
        .with_tls(acceptor)
        .with_input_handler(TestInputHandler)
        .with_display_handler(TestDisplay {
            rx: Arc::new(Mutex::new(display_rx)),
        })
        .with_connection_handler(Some(Box::new(HookRecorder {
            accept: Arc::clone(&accept),
            info: Arc::clone(&fired),
        })))
        .with_connection_policy(policy)
        .build();
    server.set_credentials(Some(server::Credentials {
        username: USERNAME.into(),
        password: PASSWORD.into(),
        domain: None,
    }));
    let ev = server.event_sender().clone();

    let local = tokio::task::LocalSet::new();
    Box::pin(local.run_until(async move {
        let server_task = tokio::task::spawn_local(async move {
            let _ = Box::pin(server.run()).await;
        });

        let (tx, rx) = oneshot::channel();
        ev.send(ServerEvent::GetLocalAddr(tx)).unwrap();
        let server_addr = rx.await.unwrap().unwrap();
        let tcp_stream = TcpStream::connect(server_addr).await.expect("TCP connect");
        let client_addr = tcp_stream.local_addr().expect("local_addr");
        let mut framed = ironrdp_tokio::TokioFramed::new(tcp_stream);
        let mut connector = connector::ClientConnector::new(default_client_config(), client_addr);
        let should_upgrade = ironrdp_async::connect_begin(&mut framed, &mut connector)
            .await
            .expect("begin connection");
        let initial_stream = framed.into_inner_no_leftover();
        let (upgraded_stream, tls_cert) = ironrdp_tls::upgrade_with_certificate_validation(
            initial_stream,
            "localhost",
            ironrdp_tls::CertificateValidation::DangerouslyAcceptInvalidCertificate,
        )
        .await
        .expect("TLS upgrade");
        let upgraded = ironrdp_tokio::mark_as_upgraded(should_upgrade, &mut connector);
        let mut upgraded_framed = ironrdp_tokio::TokioFramed::new(upgraded_stream);
        let server_public_key =
            ironrdp_tls::extract_tls_server_public_key(&tls_cert).expect("extract server public key");
        let _connection_result = ironrdp_async::connect_finalize(
            upgraded,
            connector,
            &mut upgraded_framed,
            &mut ironrdp_tokio::reqwest::ReqwestNetworkClient::new(),
            "localhost".into(),
            server_public_key.to_owned(),
            None,
        )
        .await
        .expect("finalize connection");

        // The client is now active, so the server has finished its side of
        // the handshake. Keep the connection open and give the hook a
        // generous window to land.
        let deadline = Instant::now() + Duration::from_secs(3);
        while !fired.load(Ordering::SeqCst) && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let result = (accept.load(Ordering::SeqCst), fired.load(Ordering::SeqCst));

        drop(upgraded_framed);
        server_task.abort();
        result
    }))
    .await
}

#[tokio::test]
async fn connection_handler_hooks_reach_handler_under_queue() {
    let (accept, info) = Box::pin(hooks_reaching_handler_under(server::ConnectionPolicy::Queue)).await;
    assert!(accept && info, "Queue: on_accept={accept}, on_connection_info={info}");
}

#[tokio::test]
async fn connection_handler_hooks_reach_handler_under_preempt() {
    let (accept, info) = Box::pin(hooks_reaching_handler_under(server::ConnectionPolicy::Preempt)).await;
    assert!(accept && info, "Preempt: on_accept={accept}, on_connection_info={info}");
}

#[tokio::test]
async fn connection_handler_hooks_reach_handler_under_reject() {
    let (accept, info) = Box::pin(hooks_reaching_handler_under(server::ConnectionPolicy::Reject)).await;
    assert!(accept && info, "Reject: on_accept={accept}, on_connection_info={info}");
}

cargo test -p ironrdp-testsuite-extra --test integration_tests_extra connection_handler_hooks_reach_handler

Root cause

The Preempt arm of RdpServer::run takes the handler out of self before it
builds the live connection, because that connection borrows &mut self for the
whole race while a candidate's on_accept needs the handler at the same time
(crates/ironrdp-server/src/server.rs, line numbers on d2bb7376):

let handler = self.connection_handler.take(); // :2337

It is only put back after the race (:2490). Both race arms —
run_connection_inner and serve_negotiated — go through
finalize_negotiated → accept_finalize → client_accepted, which fires the
hook only when the handler is present:

if !result.reactivation
    && let Some(ref mut handler) = self.connection_handler // :3470 — None during the race

That is the only on_connection_info call site, and there is no else, so
nothing is logged. on_accept survives because it is called at :2315, before
the take. The take dates from #1476; #1913 moved it into the
ConnectionPolicy::Preempt arm.

Impact

• on_connection_info (feat(server)!: expose per-connection keyboard metadata via ConnectionHandler #1691) is dead under Preempt for any handler relying on
  it for audit, metrics or per-session setup.
• Any future hook fired mid-connection through self.connection_handler inherits
  the same problem — including the auth-outcome hook discussed in ConnectionHandler: observation-only auth-outcome hook, or route through CredentialValidator? #1484, which is
  how I ran into this.
• No existing test asserts that on_connection_info fires at all, under any
  policy, which is how it got through.
• Timing: the pending release PR (chore(release): prepare for publishing #1880) ships feat(server): let an authenticated connection preempt an existing session #1476 and feat(server): add ConnectionPolicy to reject a second connection #1913 in
  ironrdp-server 0.14.0, so this goes out to crates.io unless it's fixed first.
  And feat(server): default ConnectionPolicy to Preempt under Hybrid #1934 proposes making preemption the default, at which point it would hit
  every embedder that serves through RdpServer::run with a handler, not only
  those who opt in. (Embedders calling run_connection directly are unaffected —
  the take only exists in run's accept loop.)

Possible fix

Share the handler rather than taking it, so the candidate's on_accept and the
live connection's hooks can both reach it.

Prior art: macrdp's fork hit exactly this. Its first cut took the handler for the
race and silently broke its auth-audit hooks for every connection — caught only
because an integration test saw zero auth records — and it was fixed by sharing
the handler as Rc<RefCell<Box<dyn ConnectionHandler>>>.

That shape should carry over without a new constraint: RdpServer is already
!Send (it holds non-Send factories such as SoundServerFactory,
CliprdrServerFactory and RdpeiServerFactory), so an Rc field doesn't take
anything away from embedders. And because every ConnectionHandler method is
synchronous, a borrow only needs to live for the duration of the call, never
across an .await; with the server confined to one thread, the candidate's
on_accept and the live connection's hooks can't hold it at the same time.
Arc<std::sync::Mutex<…>> would work too if you'd rather not tie this to the
server staying !Send.

Happy to open a PR with that and the regression tests above if the direction
works for you.

cc Anton Mostovoy (@antonmos) Marynych Oleksandr (@maryny4), since this sits in the code from #1476 and #1913.