From b011ebfcf467fc318a35b7e06525fba62169a333 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Tue, 30 Jun 2026 22:01:38 -0700 Subject: [PATCH 01/15] Wire broker readiness notifications Add a paired host-side notification serving path that preserves control request/response sequencing while sending event readiness notifications over the notification channel. Add a broker-local notification receiver and a Unix-socket runtime test covering host emission through local consumption. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 198 +++++++++++++++--- litebox_broker_local/src/lib.rs | 66 +++++- .../tests/notification_runtime.rs | 60 ++++++ 3 files changed, 292 insertions(+), 32 deletions(-) create mode 100644 litebox_broker_userland/tests/notification_runtime.rs diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index 20fd3cb3d0..84f9521d33 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -14,11 +14,14 @@ extern crate std; use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential}; use litebox_broker_protocol::BROKER_PROTOCOL_VERSION; -use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential}; +use litebox_broker_protocol::channel::{ + HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential, +}; use litebox_broker_protocol::error::ErrorCode; use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse}; use litebox_broker_protocol::message::{ - BrokerHandshakeResponse, BrokerRequest, BrokerResponse, EventRequest, EventResponse, + BrokerHandshakeResponse, BrokerNotification, BrokerRequest, BrokerResponse, + EventReadinessNotification, EventRequest, EventResponse, }; mod error; @@ -32,6 +35,39 @@ pub fn serve_connection( ) -> Result where Channel: HostControlChannel, +{ + serve_connection_inner(core, channel, |_notification| Ok(())) +} + +/// Authenticates, negotiates, and serves one broker association over paired +/// control and notification channels. +/// +/// The deployment must bind both channels to the same authenticated peer +/// association. Active requests and responses remain on the control channel; +/// broker-initiated readiness wakeups are sent on the notification channel. +pub fn serve_connection_with_notifications( + core: &BrokerCore, + control_channel: &mut ControlChannel, + notification_channel: &mut NotificationChannel, +) -> Result +where + ControlChannel: HostControlChannel, + NotificationChannel: HostNotificationChannel, +{ + serve_connection_inner(core, control_channel, |notification| { + notification_channel.send_notification(&BrokerNotification::EventReadiness(notification)) + }) +} + +fn serve_connection_inner( + core: &BrokerCore, + channel: &mut Channel, + mut notify_event_readiness: NotifyEventReadiness, +) -> Result +where + Channel: HostControlChannel, + NotifyEventReadiness: + FnMut(EventReadinessNotification) -> core::result::Result<(), Channel::Error>, { let peer_credential = channel .peer_credential() @@ -77,15 +113,18 @@ where } } - serve_request_loop(channel, &session) + serve_request_loop(channel, &session, &mut notify_event_readiness) } -fn serve_request_loop( +fn serve_request_loop( channel: &mut Channel, session: &BrokerSession, + notify_event_readiness: &mut NotifyEventReadiness, ) -> Result where Channel: HostControlChannel, + NotifyEventReadiness: + FnMut(EventReadinessNotification) -> core::result::Result<(), Channel::Error>, { loop { let request = match channel.recv_request().map_err(BrokerHostError::Channel)? { @@ -99,60 +138,89 @@ where HostReceive::PeerClosed => break, }; - let response = handle_request(session, request); + let handled = handle_request(session, request); channel - .send_response(&response) + .send_response(&handled.response) .map_err(BrokerHostError::Channel)?; + if let Some(notification) = handled.event_readiness_notification { + notify_event_readiness(notification).map_err(BrokerHostError::Channel)?; + } } Ok(ConnectionTermination::PeerClosed) } -fn handle_request(session: &BrokerSession, request: BrokerRequest) -> BrokerResponse { +fn handle_request(session: &BrokerSession, request: BrokerRequest) -> HandledRequest { match request { BrokerRequest::CloseObject(handle) => match session.close_object_reference(handle) { - Ok(()) => BrokerResponse::ObjectClosed, - Err(error) => BrokerResponse::Error(error.into()), + Ok(()) => HandledRequest::response(BrokerResponse::ObjectClosed), + Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), }, BrokerRequest::Event(request) => handle_event_request(session, request), } } -fn handle_event_request(session: &BrokerSession, request: EventRequest) -> BrokerResponse { +fn handle_event_request(session: &BrokerSession, request: EventRequest) -> HandledRequest { match request { EventRequest::Create(request) => { match litebox_broker_core::event::create(session, request.initial_count) { - Ok(handle) => { - BrokerResponse::Event(EventResponse::Create(CreateEventResponse { handle })) - } - Err(error) => BrokerResponse::Error(error.into()), + Ok(handle) => HandledRequest::response(BrokerResponse::Event( + EventResponse::Create(CreateEventResponse { handle }), + )), + Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), } } EventRequest::Wait(request) => { match litebox_broker_core::event::wait(session, request.handle) { - Ok(readiness) => { - BrokerResponse::Event(EventResponse::Wait(WaitEventResponse { readiness })) - } - Err(error) => BrokerResponse::Error(error.into()), + Ok(readiness) => HandledRequest::response(BrokerResponse::Event( + EventResponse::Wait(WaitEventResponse { readiness }), + )), + Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), } } EventRequest::Add(request) => { match litebox_broker_core::event::add(session, request.handle, request.value) { - Ok(readiness) => { - BrokerResponse::Event(EventResponse::Add(AddEventResponse { readiness })) - } - Err(error) => BrokerResponse::Error(error.into()), + Ok(readiness) => HandledRequest { + response: BrokerResponse::Event(EventResponse::Add(AddEventResponse { + readiness, + })), + event_readiness_notification: Some(EventReadinessNotification { + handle: request.handle, + readiness, + }), + }, + Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), } } EventRequest::Consume(request) => { match litebox_broker_core::event::consume(session, request.handle, request.mode) { - Ok(consumption) => BrokerResponse::Event(EventResponse::Consume(consumption)), - Err(error) => BrokerResponse::Error(error.into()), + Ok(consumption) => HandledRequest { + event_readiness_notification: Some(EventReadinessNotification { + handle: request.handle, + readiness: consumption.readiness, + }), + response: BrokerResponse::Event(EventResponse::Consume(consumption)), + }, + Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), } } } } +struct HandledRequest { + response: BrokerResponse, + event_readiness_notification: Option, +} + +impl HandledRequest { + fn response(response: BrokerResponse) -> Self { + Self { + response, + event_readiness_notification: None, + } + } +} + /// Terminal outcome after processing one broker connection. #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[non_exhaustive] @@ -183,6 +251,7 @@ mod tests { serve_connection_rejects_active_request_before_negotiation(&broker); serve_connection_rejects_handshake_request_after_negotiation(&broker); serve_connection_returns_channel_error_when_response_send_fails(&broker); + serve_request_loop_sends_event_readiness_notifications(&broker); active_request_closes_object_reference(&broker); } @@ -303,6 +372,59 @@ mod tests { assert!(channel.handshake_responses.is_empty()); } + fn serve_request_loop_sends_event_readiness_notifications(broker: &BrokerCore) { + let session = broker + .create_session(CallerCredential::Unauthenticated) + .unwrap(); + let handle = litebox_broker_core::event::create(&session, 0).unwrap(); + let mut channel = FakeHostControlChannel::new( + std::vec::Vec::new(), + std::vec::Vec::from([ + Ok(HostReceive::Message(BrokerRequest::Event( + EventRequest::Add(litebox_broker_protocol::event::AddEventRequest { + handle, + value: 1, + }), + ))), + Ok(HostReceive::Message(BrokerRequest::Event( + EventRequest::Consume(litebox_broker_protocol::event::ConsumeEventRequest { + handle, + mode: litebox_broker_protocol::event::EventConsumeMode::One, + }), + ))), + Ok(HostReceive::PeerClosed), + ]), + ); + let mut notifications = FakeHostNotificationChannel::default(); + + assert_eq!( + serve_request_loop(&mut channel, &session, &mut |notification| { + notifications.send_notification(&BrokerNotification::EventReadiness(notification)) + }) + .unwrap(), + ConnectionTermination::PeerClosed + ); + assert_eq!( + notifications.notifications, + [ + BrokerNotification::EventReadiness(EventReadinessNotification { + handle, + readiness: litebox_broker_protocol::event::ReadinessState { + read_ready: true, + write_ready: true, + }, + }), + BrokerNotification::EventReadiness(EventReadinessNotification { + handle, + readiness: litebox_broker_protocol::event::ReadinessState { + read_ready: false, + write_ready: true, + }, + }), + ] + ); + } + fn active_request_closes_object_reference(broker: &BrokerCore) { let session = broker .create_session(CallerCredential::Unauthenticated) @@ -312,28 +434,31 @@ mod tests { BrokerRequest::Event(EventRequest::Create(CreateEventRequest { initial_count: 0, })), - ); + ) + .response; let BrokerResponse::Event(EventResponse::Create(response)) = response else { panic!("unexpected create response: {response:?}"); }; let handle = response.handle; assert_eq!( - handle_request(&session, BrokerRequest::CloseObject(handle)), + handle_request(&session, BrokerRequest::CloseObject(handle)).response, BrokerResponse::ObjectClosed ); assert_eq!( handle_request( &session, BrokerRequest::Event(EventRequest::Wait(WaitEventRequest { handle })) - ), + ) + .response, BrokerResponse::Error(ErrorCode::UnknownObject) ); assert_eq!( handle_request( &session, BrokerRequest::CloseObject(ObjectHandle(handle.0 + 1)) - ), + ) + .response, BrokerResponse::Error(ErrorCode::UnknownObject) ); } @@ -413,4 +538,21 @@ mod tests { Ok(()) } } + + #[derive(Default)] + struct FakeHostNotificationChannel { + notifications: std::vec::Vec, + } + + impl HostNotificationChannel for FakeHostNotificationChannel { + type Error = (); + + fn send_notification( + &mut self, + notification: &BrokerNotification, + ) -> core::result::Result<(), Self::Error> { + self.notifications.push(notification.clone()); + Ok(()) + } + } } diff --git a/litebox_broker_local/src/lib.rs b/litebox_broker_local/src/lib.rs index f90e481af1..8c2ceecd5c 100644 --- a/litebox_broker_local/src/lib.rs +++ b/litebox_broker_local/src/lib.rs @@ -1,11 +1,13 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT license. -//! Typed broker-local control adapter for broker requests. +//! Typed broker-local adapters for broker requests and notifications. //! //! The local control adapter owns request/response sequencing but does not own a channel. //! Userland, kernel, or ring-buffer deployments can provide channels by //! implementing [`litebox_broker_protocol::channel::LocalControlChannel`]. +//! Notification receive adapters are intentionally separate so active control +//! requests remain strictly paired with their responses. #![no_std] @@ -15,10 +17,11 @@ extern crate std; mod error; mod event; -use litebox_broker_protocol::channel::LocalControlChannel; +use litebox_broker_protocol::channel::{LocalControlChannel, LocalNotificationChannel}; use litebox_broker_protocol::error::ErrorCode; use litebox_broker_protocol::message::{ - BrokerHandshakeRequest, BrokerHandshakeResponse, BrokerRequest, BrokerResponse, + BrokerHandshakeRequest, BrokerHandshakeResponse, BrokerNotification, BrokerRequest, + BrokerResponse, }; use litebox_broker_protocol::{BROKER_PROTOCOL_VERSION, ObjectHandle}; @@ -29,6 +32,11 @@ pub struct BrokerLocal { channel: Channel, } +/// Broker-local receive adapter for broker-initiated asynchronous notifications. +pub struct BrokerNotifications { + channel: Channel, +} + impl BrokerLocal { /// Negotiates the broker protocol over an already-connected control channel. /// @@ -124,14 +132,33 @@ impl BrokerLocal { } } +impl BrokerNotifications { + /// Creates a notification receiver from an already-associated notification channel. + pub const fn new(channel: Channel) -> Self { + Self { channel } + } + + /// Receives the next broker notification. + /// + /// Returns `Ok(None)` when the broker closed the notification channel cleanly. + pub fn recv_notification(&mut self) -> Result, Channel::Error> { + self.channel + .recv_notification() + .map_err(BrokerLocalError::Channel) + } +} + #[cfg(test)] mod tests { use super::*; use core::convert::Infallible; use litebox_broker_protocol::ObjectHandle; use litebox_broker_protocol::ProtocolVersion; + use litebox_broker_protocol::channel::LocalNotificationChannel; use litebox_broker_protocol::event::{CreateEventRequest, CreateEventResponse}; - use litebox_broker_protocol::message::{EventRequest, EventResponse}; + use litebox_broker_protocol::message::{ + EventReadinessNotification, EventRequest, EventResponse, + }; #[test] fn negotiate_returns_active_local_connection() { @@ -219,6 +246,23 @@ mod tests { let _ = BrokerLocal::negotiate(channel); } + #[test] + fn notification_receiver_returns_broker_notifications() { + let notification = BrokerNotification::EventReadiness(EventReadinessNotification { + handle: ObjectHandle(7), + readiness: litebox_broker_protocol::event::ReadinessState { + read_ready: true, + write_ready: false, + }, + }); + let mut receiver = BrokerNotifications::new(FakeNotificationChannel { + notification: Some(notification.clone()), + }); + + assert_eq!(receiver.recv_notification().unwrap(), Some(notification)); + assert_eq!(receiver.recv_notification().unwrap(), None); + } + #[test] fn negotiate_rejects_broker_unsupported_version_response() { let broker_protocol_version = ProtocolVersion(BROKER_PROTOCOL_VERSION.0 + 1); @@ -296,4 +340,18 @@ mod tests { Ok(self.response.take()) } } + + struct FakeNotificationChannel { + notification: Option, + } + + impl LocalNotificationChannel for FakeNotificationChannel { + type Error = Infallible; + + fn recv_notification( + &mut self, + ) -> core::result::Result, Self::Error> { + Ok(self.notification.take()) + } + } } diff --git a/litebox_broker_userland/tests/notification_runtime.rs b/litebox_broker_userland/tests/notification_runtime.rs new file mode 100644 index 0000000000..97a15c21c8 --- /dev/null +++ b/litebox_broker_userland/tests/notification_runtime.rs @@ -0,0 +1,60 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +use std::os::unix::net::UnixStream; +use std::time::Duration; + +use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights}; +use litebox_broker_host::{ConnectionTermination, serve_connection_with_notifications}; +use litebox_broker_local::{BrokerLocal, BrokerNotifications}; +use litebox_broker_protocol::event::ReadinessState; +use litebox_broker_protocol::message::{BrokerNotification, EventReadinessNotification}; +use litebox_broker_transport::unix_socket::{ + UnixStreamHostControlChannel, UnixStreamHostNotificationChannel, UnixStreamLocalControlChannel, + UnixStreamLocalNotificationChannel, +}; + +#[test] +fn host_sends_readiness_notifications_over_paired_userland_channel() { + let broker = BrokerCore::new(PolicyEngine::with_unauthenticated_rights( + PrincipalRights::all(), + )) + .unwrap(); + let (local_control, host_control) = UnixStream::pair().unwrap(); + let (local_notification, host_notification) = UnixStream::pair().unwrap(); + local_notification + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + + let host_thread = std::thread::spawn(move || { + let mut control = UnixStreamHostControlChannel::from_accepted(host_control); + let mut notification = UnixStreamHostNotificationChannel::from_accepted(host_notification); + serve_connection_with_notifications(&broker, &mut control, &mut notification) + }); + + let mut local = + BrokerLocal::negotiate(UnixStreamLocalControlChannel::from_connected(local_control)) + .unwrap(); + let mut notifications = BrokerNotifications::new( + UnixStreamLocalNotificationChannel::from_connected(local_notification), + ); + + let handle = local.create_event_with_count(0).unwrap(); + let readiness = ReadinessState { + read_ready: true, + write_ready: true, + }; + assert_eq!(local.add_event(handle, 1).unwrap(), readiness); + assert_eq!( + notifications.recv_notification().unwrap(), + Some(BrokerNotification::EventReadiness( + EventReadinessNotification { handle, readiness } + )) + ); + + drop(local); + assert_eq!( + host_thread.join().unwrap().unwrap(), + ConnectionTermination::PeerClosed + ); +} From 1f2007e11f2d9ff789c5762d9435d4d399e232a8 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 07:09:52 -0700 Subject: [PATCH 02/15] Clarify broker host channel generic names Rename generic control-channel parameters from Channel to ControlChannel so the host adapter remains consistent now that paired notification channels are part of the API. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 49 ++++++++++++++++++---------------- 1 file changed, 26 insertions(+), 23 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index 84f9521d33..1ddd6341aa 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -29,14 +29,14 @@ mod error; pub use error::{BrokerHostError, Result}; /// Authenticates, negotiates, and serves one broker connection over the control channel. -pub fn serve_connection( +pub fn serve_connection( core: &BrokerCore, - channel: &mut Channel, -) -> Result + control_channel: &mut ControlChannel, +) -> Result where - Channel: HostControlChannel, + ControlChannel: HostControlChannel, { - serve_connection_inner(core, channel, |_notification| Ok(())) + serve_connection_inner(core, control_channel, |_notification| Ok(())) } /// Authenticates, negotiates, and serves one broker association over paired @@ -59,17 +59,17 @@ where }) } -fn serve_connection_inner( +fn serve_connection_inner( core: &BrokerCore, - channel: &mut Channel, + control_channel: &mut ControlChannel, mut notify_event_readiness: NotifyEventReadiness, -) -> Result +) -> Result where - Channel: HostControlChannel, + ControlChannel: HostControlChannel, NotifyEventReadiness: - FnMut(EventReadinessNotification) -> core::result::Result<(), Channel::Error>, + FnMut(EventReadinessNotification) -> core::result::Result<(), ControlChannel::Error>, { - let peer_credential = channel + let peer_credential = control_channel .peer_credential() .map_err(BrokerHostError::Channel)?; let caller_credential = match peer_credential { @@ -79,13 +79,13 @@ where let session = core.create_session(caller_credential)?; loop { - let request = match channel + let request = match control_channel .recv_handshake_request() .map_err(BrokerHostError::Channel)? { HostReceive::Message(request) => request, HostReceive::ProtocolViolation => { - channel + control_channel .send_handshake_response(&BrokerHandshakeResponse::Error( ErrorCode::ProtocolState, )) @@ -105,7 +105,7 @@ where broker_protocol_version: BROKER_PROTOCOL_VERSION, } }; - channel + control_channel .send_handshake_response(&response) .map_err(BrokerHostError::Channel)?; if negotiated { @@ -113,24 +113,27 @@ where } } - serve_request_loop(channel, &session, &mut notify_event_readiness) + serve_request_loop(control_channel, &session, &mut notify_event_readiness) } -fn serve_request_loop( - channel: &mut Channel, +fn serve_request_loop( + control_channel: &mut ControlChannel, session: &BrokerSession, notify_event_readiness: &mut NotifyEventReadiness, -) -> Result +) -> Result where - Channel: HostControlChannel, + ControlChannel: HostControlChannel, NotifyEventReadiness: - FnMut(EventReadinessNotification) -> core::result::Result<(), Channel::Error>, + FnMut(EventReadinessNotification) -> core::result::Result<(), ControlChannel::Error>, { loop { - let request = match channel.recv_request().map_err(BrokerHostError::Channel)? { + let request = match control_channel + .recv_request() + .map_err(BrokerHostError::Channel)? + { HostReceive::Message(request) => request, HostReceive::ProtocolViolation => { - channel + control_channel .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState)) .map_err(BrokerHostError::Channel)?; return Ok(ConnectionTermination::ProtocolViolation); @@ -139,7 +142,7 @@ where }; let handled = handle_request(session, request); - channel + control_channel .send_response(&handled.response) .map_err(BrokerHostError::Channel)?; if let Some(notification) = handled.event_readiness_notification { From f4c14a2c23055c95af2c6b9793fbbb1e6eb9785d Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 07:15:38 -0700 Subject: [PATCH 03/15] Require broker notification channel Remove the control-only host serving path and route broker userland and runner integration through paired control and notification sockets. Keep the runner-side notification channel open with a receiver loop so host readiness notifications have a real local endpoint before eventfd blocking dispatch is added. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 59 ++++------ litebox_broker_userland/src/main.rs | 30 ++++-- .../tests/userland_broker.rs | 69 +++++++++--- litebox_runner_linux_userland/src/broker.rs | 102 +++++++++++++++--- litebox_runner_linux_userland/src/lib.rs | 15 ++- litebox_runner_linux_userland/tests/run.rs | 83 ++++++++++---- 6 files changed, 253 insertions(+), 105 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index 1ddd6341aa..424b61dfea 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -28,17 +28,6 @@ mod error; pub use error::{BrokerHostError, Result}; -/// Authenticates, negotiates, and serves one broker connection over the control channel. -pub fn serve_connection( - core: &BrokerCore, - control_channel: &mut ControlChannel, -) -> Result -where - ControlChannel: HostControlChannel, -{ - serve_connection_inner(core, control_channel, |_notification| Ok(())) -} - /// Authenticates, negotiates, and serves one broker association over paired /// control and notification channels. /// @@ -53,21 +42,6 @@ pub fn serve_connection_with_notifications( where ControlChannel: HostControlChannel, NotificationChannel: HostNotificationChannel, -{ - serve_connection_inner(core, control_channel, |notification| { - notification_channel.send_notification(&BrokerNotification::EventReadiness(notification)) - }) -} - -fn serve_connection_inner( - core: &BrokerCore, - control_channel: &mut ControlChannel, - mut notify_event_readiness: NotifyEventReadiness, -) -> Result -where - ControlChannel: HostControlChannel, - NotifyEventReadiness: - FnMut(EventReadinessNotification) -> core::result::Result<(), ControlChannel::Error>, { let peer_credential = control_channel .peer_credential() @@ -113,18 +87,17 @@ where } } - serve_request_loop(control_channel, &session, &mut notify_event_readiness) + serve_request_loop(control_channel, notification_channel, &session) } -fn serve_request_loop( +fn serve_request_loop( control_channel: &mut ControlChannel, + notification_channel: &mut NotificationChannel, session: &BrokerSession, - notify_event_readiness: &mut NotifyEventReadiness, ) -> Result where ControlChannel: HostControlChannel, - NotifyEventReadiness: - FnMut(EventReadinessNotification) -> core::result::Result<(), ControlChannel::Error>, + NotificationChannel: HostNotificationChannel, { loop { let request = match control_channel @@ -146,7 +119,9 @@ where .send_response(&handled.response) .map_err(BrokerHostError::Channel)?; if let Some(notification) = handled.event_readiness_notification { - notify_event_readiness(notification).map_err(BrokerHostError::Channel)?; + notification_channel + .send_notification(&BrokerNotification::EventReadiness(notification)) + .map_err(BrokerHostError::Channel)?; } } @@ -270,9 +245,10 @@ mod tests { Ok(HostReceive::PeerClosed), ]), ); + let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection(broker, &mut channel).unwrap(), + serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::PeerClosed ); assert_eq!( @@ -300,9 +276,10 @@ mod tests { ]), std::vec::Vec::from([Ok(HostReceive::PeerClosed)]), ); + let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection(broker, &mut channel).unwrap(), + serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::PeerClosed ); assert_eq!( @@ -323,9 +300,10 @@ mod tests { std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]), std::vec::Vec::new(), ); + let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection(broker, &mut channel).unwrap(), + serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::ProtocolViolation ); assert_eq!( @@ -342,9 +320,10 @@ mod tests { }))]), std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]), ); + let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection(broker, &mut channel).unwrap(), + serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::ProtocolViolation ); assert_eq!( @@ -367,8 +346,9 @@ mod tests { std::vec::Vec::new(), ); channel.send_error = true; + let mut notifications = FakeHostNotificationChannel::default(); - match serve_connection(broker, &mut channel) { + match serve_connection_with_notifications(broker, &mut channel, &mut notifications) { Err(BrokerHostError::Channel(())) => {} result => panic!("unexpected serve result: {result:?}"), } @@ -401,10 +381,7 @@ mod tests { let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_request_loop(&mut channel, &session, &mut |notification| { - notifications.send_notification(&BrokerNotification::EventReadiness(notification)) - }) - .unwrap(), + serve_request_loop(&mut channel, &mut notifications, &session).unwrap(), ConnectionTermination::PeerClosed ); assert_eq!( diff --git a/litebox_broker_userland/src/main.rs b/litebox_broker_userland/src/main.rs index b6c2ba559b..0c9b1bf070 100644 --- a/litebox_broker_userland/src/main.rs +++ b/litebox_broker_userland/src/main.rs @@ -9,8 +9,10 @@ use std::process::Command; use clap::Parser; use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights}; -use litebox_broker_host::serve_connection; -use litebox_broker_transport::unix_socket::UnixStreamHostControlChannel; +use litebox_broker_host::serve_connection_with_notifications; +use litebox_broker_transport::unix_socket::{ + UnixStreamHostControlChannel, UnixStreamHostNotificationChannel, +}; #[derive(Parser, Debug)] struct CliArgs { @@ -27,8 +29,10 @@ fn main() -> Result<(), Box> { let socket_dir = tempfile::Builder::new() .prefix("litebox-broker-userland-") .tempdir()?; - let socket_path = socket_dir.path().join("broker.sock"); - let listener = UnixListener::bind(&socket_path)?; + let control_socket_path = socket_dir.path().join("broker.sock"); + let notification_socket_path = socket_dir.path().join("broker-notification.sock"); + let control_listener = UnixListener::bind(&control_socket_path)?; + let notification_listener = UnixListener::bind(¬ification_socket_path)?; let broker = BrokerCore::new(PolicyEngine::with_unauthenticated_rights( PrincipalRights::all(), ))?; @@ -37,7 +41,9 @@ fn main() -> Result<(), Box> { runner_command .arg("--unstable") .arg("--broker-socket") - .arg(&socket_path) + .arg(&control_socket_path) + .arg("--broker-notification-socket") + .arg(¬ification_socket_path) .args(&args.runner_arguments); let mut runner = runner_command.spawn()?; let _runner_waiter = std::thread::spawn(move || { @@ -47,13 +53,21 @@ fn main() -> Result<(), Box> { }); loop { - let (stream, _) = listener.accept()?; + let (control_stream, _) = control_listener.accept()?; + let (notification_stream, _) = notification_listener.accept()?; let broker = broker.clone(); if let Err(error) = std::thread::Builder::new() .name("litebox-broker-connection".to_owned()) .spawn(move || { - let mut channel = UnixStreamHostControlChannel::from_accepted(stream); - if let Err(error) = serve_connection(&broker, &mut channel) { + let mut control_channel = + UnixStreamHostControlChannel::from_accepted(control_stream); + let mut notification_channel = + UnixStreamHostNotificationChannel::from_accepted(notification_stream); + if let Err(error) = serve_connection_with_notifications( + &broker, + &mut control_channel, + &mut notification_channel, + ) { eprintln!("failed to serve broker connection: {error}"); } }) diff --git a/litebox_broker_userland/tests/userland_broker.rs b/litebox_broker_userland/tests/userland_broker.rs index 6bc055aed4..3f98489a2b 100644 --- a/litebox_broker_userland/tests/userland_broker.rs +++ b/litebox_broker_userland/tests/userland_broker.rs @@ -8,9 +8,12 @@ use std::path::Path; use std::process::{Child, Command}; use std::time::{Duration, Instant}; -use litebox_broker_local::BrokerLocal; +use litebox_broker_local::{BrokerLocal, BrokerNotifications}; use litebox_broker_protocol::event::ReadinessState; -use litebox_broker_transport::unix_socket::UnixStreamLocalControlChannel; +use litebox_broker_protocol::message::{BrokerNotification, EventReadinessNotification}; +use litebox_broker_transport::unix_socket::{ + UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel, +}; const RUNNER_ARGUMENT: &str = "broker-userland-test-runner"; @@ -30,10 +33,11 @@ fn run_parent_test() { // This custom-harness integration test uses its own executable as the broker's // runner. Cargo starts this executable without broker args, so it runs the // parent path here. The broker then starts the same executable with the real - // runner argv (`--unstable --broker-socket `), which runs `run_fake_runner`. - // After the fake runner finishes its broker requests, it terminates the broker - // parent process; this lets the test exercise the long-running broker without a - // test-only shutdown path. + // runner argv (`--unstable --broker-socket + // --broker-notification-socket `), which runs `run_fake_runner`. After + // the fake runner finishes its broker requests, it terminates the broker + // parent process; this lets the test exercise the long-running broker + // without a test-only shutdown path. let mut broker = ChildGuard { child: Command::new(env!("CARGO_BIN_EXE_litebox-broker-userland")) .arg("--runner") @@ -65,13 +69,21 @@ fn run_fake_runner(args: &[OsString]) { ); assert_eq!( args.get(3).map(OsString::as_os_str), + Some(OsStr::new("--broker-notification-socket")) + ); + assert_eq!( + args.get(5).map(OsString::as_os_str), Some(OsStr::new(RUNNER_ARGUMENT)) ); - assert_eq!(args.len(), 4, "unexpected runner arguments: {args:?}"); + assert_eq!(args.len(), 6, "unexpected runner arguments: {args:?}"); - let socket_path = args.get(2).unwrap(); - let channel = connect_with_retry(Path::new(socket_path)).unwrap(); - let mut local = BrokerLocal::negotiate(channel).unwrap(); + let control_socket_path = args.get(2).unwrap(); + let notification_socket_path = args.get(4).unwrap(); + let control_channel = connect_control_with_retry(Path::new(control_socket_path)).unwrap(); + let notification_channel = + connect_notification_with_retry(Path::new(notification_socket_path)).unwrap(); + let mut local = BrokerLocal::negotiate(control_channel).unwrap(); + let mut notifications = BrokerNotifications::new(notification_channel); let handle = local.create_event_with_count(0).unwrap(); assert_eq!( @@ -82,12 +94,16 @@ fn run_fake_runner(args: &[OsString]) { } ); + let readiness = ReadinessState { + read_ready: true, + write_ready: true, + }; + assert_eq!(local.add_event(handle, 1).unwrap(), readiness); assert_eq!( - local.add_event(handle, 1).unwrap(), - ReadinessState { - read_ready: true, - write_ready: true, - } + notifications.recv_notification().unwrap(), + Some(BrokerNotification::EventReadiness( + EventReadinessNotification { handle, readiness } + )) ); assert_eq!( @@ -97,6 +113,7 @@ fn run_fake_runner(args: &[OsString]) { write_ready: true, } ); + drop(notifications); drop(local); // SAFETY: `getppid` takes no pointer arguments and has no Rust-side aliasing requirements. @@ -124,7 +141,7 @@ impl Drop for ChildGuard { } } -fn connect_with_retry(socket_path: &Path) -> Result { +fn connect_control_with_retry(socket_path: &Path) -> Result { let deadline = Instant::now() + Duration::from_secs(5); loop { match UnixStreamLocalControlChannel::connect_with_setup_deadline(socket_path, deadline) { @@ -141,3 +158,23 @@ fn connect_with_retry(socket_path: &Path) -> Result Result { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + match UnixStreamLocalNotificationChannel::connect(socket_path) { + Ok(channel) => return Ok(channel), + Err(error) if Instant::now() < deadline => { + if error.kind() != ErrorKind::NotFound + && error.kind() != ErrorKind::ConnectionRefused + { + return Err(error); + } + std::thread::sleep(Duration::from_millis(10)); + } + Err(error) => return Err(error), + } + } +} diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs index 05737cae24..097bda62b8 100644 --- a/litebox_runner_linux_userland/src/broker.rs +++ b/litebox_runner_linux_userland/src/broker.rs @@ -3,12 +3,15 @@ use std::{ path::Path, + thread::JoinHandle, time::{Duration, Instant}, }; -use anyhow::{Context as _, Result}; -use litebox_broker_local::BrokerLocal; -use litebox_broker_transport::unix_socket::UnixStreamLocalControlChannel; +use anyhow::{Context as _, Result, bail}; +use litebox_broker_local::{BrokerLocal, BrokerNotifications}; +use litebox_broker_transport::unix_socket::{ + UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel, +}; const SETUP_TIMEOUT: Duration = Duration::from_secs(5); const RETRY_DELAY: Duration = Duration::from_millis(20); @@ -16,12 +19,24 @@ type Local = BrokerLocal; pub(crate) struct BrokerConnection { local: Local, + _notification_thread: JoinHandle<()>, } -pub(crate) fn connect(socket_path: Option<&Path>) -> Result> { - match socket_path { - Some(path) => connect_to_endpoint(path).map(Some), - None => Ok(None), +pub(crate) fn connect( + control_socket_path: Option<&Path>, + notification_socket_path: Option<&Path>, +) -> Result> { + match (control_socket_path, notification_socket_path) { + (Some(control_path), Some(notification_path)) => { + connect_to_endpoint(control_path, notification_path).map(Some) + } + (None, None) => Ok(None), + (Some(_), None) => { + bail!("broker notification socket is required with broker control socket") + } + (None, Some(_)) => { + bail!("broker control socket is required with broker notification socket") + } } } @@ -31,23 +46,60 @@ impl BrokerConnection { } } -fn connect_to_endpoint(socket_path: &Path) -> Result { +fn connect_to_endpoint( + control_socket_path: &Path, + notification_socket_path: &Path, +) -> Result { let setup_deadline = Instant::now() + SETUP_TIMEOUT; - let local = connect_with_retry(socket_path, setup_deadline) - .with_context(|| format!("failed to connect to broker at {}", socket_path.display()))?; - Ok(BrokerConnection { local }) + let control_channel = connect_control_with_retry(control_socket_path, setup_deadline) + .with_context(|| { + format!( + "failed to connect to broker at {}", + control_socket_path.display() + ) + })?; + let notification_channel = + connect_notification_with_retry(notification_socket_path, setup_deadline).with_context( + || { + format!( + "failed to connect to broker notifications at {}", + notification_socket_path.display() + ) + }, + )?; + let local = BrokerLocal::negotiate(control_channel).context("broker negotiation failed")?; + let mut notifications = BrokerNotifications::new(notification_channel); + let notification_thread = std::thread::Builder::new() + .name("litebox-broker-notifications".to_owned()) + .spawn(move || { + loop { + match notifications.recv_notification() { + Ok(Some(_notification)) => {} + Ok(None) => break, + Err(error) => { + eprintln!("failed to receive broker notification: {error}"); + break; + } + } + } + }) + .context("failed to start broker notification receiver")?; + Ok(BrokerConnection { + local, + _notification_thread: notification_thread, + }) } -fn connect_with_retry(socket_path: &Path, setup_deadline: Instant) -> Result { +fn connect_control_with_retry( + socket_path: &Path, + setup_deadline: Instant, +) -> Result { loop { match UnixStreamLocalControlChannel::connect_with_setup_deadline( socket_path, setup_deadline, ) { - Ok(channel) => { - let local = BrokerLocal::negotiate(channel).context("broker negotiation failed")?; - return Ok(local); - } + Ok(channel) => return Ok(channel), Err(error) => { if Instant::now() >= setup_deadline { return Err(error).context("timed out connecting to broker"); @@ -58,3 +110,21 @@ fn connect_with_retry(socket_path: &Path, setup_deadline: Instant) -> Result Result { + loop { + match UnixStreamLocalNotificationChannel::connect(socket_path) { + Ok(channel) => return Ok(channel), + Err(error) => { + if Instant::now() >= setup_deadline { + return Err(error).context("timed out connecting to broker notifications"); + } + } + } + let remaining = setup_deadline.saturating_duration_since(Instant::now()); + std::thread::sleep(RETRY_DELAY.min(remaining)); + } +} diff --git a/litebox_runner_linux_userland/src/lib.rs b/litebox_runner_linux_userland/src/lib.rs index efe4406325..e01997e874 100644 --- a/litebox_runner_linux_userland/src/lib.rs +++ b/litebox_runner_linux_userland/src/lib.rs @@ -89,6 +89,16 @@ pub struct CliArgs { help_heading = "Unstable Options" )] pub broker_socket: Option, + /// Broker-supplied Unix socket path for the local notification channel. + #[arg( + long = "broker-notification-socket", + value_name = "PATH", + value_hint = clap::ValueHint::FilePath, + hide = true, + requires = "unstable", + help_heading = "Unstable Options" + )] + pub broker_notification_socket: Option, } struct MmappedFile { @@ -213,7 +223,10 @@ pub fn run(cli_args: CliArgs) -> Result<()> { } litebox_platform_multiplex::set_platform(platform); - let broker_connection = broker::connect(cli_args.broker_socket.as_deref())?; + let broker_connection = broker::connect( + cli_args.broker_socket.as_deref(), + cli_args.broker_notification_socket.as_deref(), + )?; let shim_builder = if let Some(broker_connection) = broker_connection { litebox_shim_linux::LinuxShimBuilder::new_with_litebox( diff --git a/litebox_runner_linux_userland/tests/run.rs b/litebox_runner_linux_userland/tests/run.rs index a5684a90a3..653cd6bdf4 100644 --- a/litebox_runner_linux_userland/tests/run.rs +++ b/litebox_runner_linux_userland/tests/run.rs @@ -123,8 +123,15 @@ impl Runner { } #[cfg(all(target_arch = "x86_64", target_os = "linux"))] - fn broker_socket(&mut self, socket_path: &Path) -> &mut Self { - self.command.arg("--broker-socket").arg(socket_path); + fn broker_socket( + &mut self, + control_socket_path: &Path, + notification_socket_path: &Path, + ) -> &mut Self { + self.command.arg("--broker-socket").arg(control_socket_path); + self.command + .arg("--broker-notification-socket") + .arg(notification_socket_path); self } @@ -264,7 +271,8 @@ struct TestBroker { thread: Option>, done_rx: std::sync::mpsc::Receiver<()>, close_object_count_rx: std::sync::mpsc::Receiver, - socket_path: PathBuf, + control_socket_path: PathBuf, + notification_socket_path: PathBuf, } #[cfg(all(target_arch = "x86_64", target_os = "linux"))] @@ -284,54 +292,79 @@ impl TestBroker { .expect("broker test host thread missing") .join() .expect("broker test host panicked"); - let _ = std::fs::remove_file(&self.socket_path); + let _ = std::fs::remove_file(&self.control_socket_path); + let _ = std::fs::remove_file(&self.notification_socket_path); } } #[cfg(all(target_arch = "x86_64", target_os = "linux"))] impl Drop for TestBroker { fn drop(&mut self) { - let _ = std::fs::remove_file(&self.socket_path); + let _ = std::fs::remove_file(&self.control_socket_path); + let _ = std::fs::remove_file(&self.notification_socket_path); } } #[cfg(all(target_arch = "x86_64", target_os = "linux"))] fn spawn_test_broker( - socket_path: &Path, + control_socket_path: &Path, + notification_socket_path: &Path, policy: litebox_broker_core::PolicyEngine, connection_count: usize, ) -> TestBroker { - let _ = std::fs::remove_file(socket_path); + let _ = std::fs::remove_file(control_socket_path); + let _ = std::fs::remove_file(notification_socket_path); let (ready_tx, ready_rx) = std::sync::mpsc::channel(); let (done_tx, done_rx) = std::sync::mpsc::channel(); let (close_object_count_tx, close_object_count_rx) = std::sync::mpsc::channel(); - let server_socket_path = socket_path.to_path_buf(); - let cleanup_socket_path = socket_path.to_path_buf(); + let server_control_socket_path = control_socket_path.to_path_buf(); + let server_notification_socket_path = notification_socket_path.to_path_buf(); + let cleanup_control_socket_path = control_socket_path.to_path_buf(); + let cleanup_notification_socket_path = notification_socket_path.to_path_buf(); let broker_thread = std::thread::spawn(move || { let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - let listener = std::os::unix::net::UnixListener::bind(&server_socket_path) - .expect("failed to bind broker test socket"); + let control_listener = + std::os::unix::net::UnixListener::bind(&server_control_socket_path) + .expect("failed to bind broker test control socket"); + let notification_listener = + std::os::unix::net::UnixListener::bind(&server_notification_socket_path) + .expect("failed to bind broker test notification socket"); let broker = litebox_broker_core::BrokerCore::new(policy).expect("failed to create broker core"); ready_tx.send(()).expect("failed to report broker ready"); for _ in 0..connection_count { - let (stream, _) = listener + let (control_stream, _) = control_listener .accept() .expect("failed to accept broker local control connection"); - stream + let (notification_stream, _) = notification_listener + .accept() + .expect("failed to accept broker local notification connection"); + control_stream .set_read_timeout(Some(BROKER_HELPER_TIMEOUT)) .expect("failed to configure broker test read timeout"); - stream + control_stream .set_write_timeout(Some(BROKER_HELPER_TIMEOUT)) .expect("failed to configure broker test write timeout"); + notification_stream + .set_read_timeout(Some(BROKER_HELPER_TIMEOUT)) + .expect("failed to configure broker notification test read timeout"); + notification_stream + .set_write_timeout(Some(BROKER_HELPER_TIMEOUT)) + .expect("failed to configure broker notification test write timeout"); let mut channel = CountingHostControlChannel { - inner: litebox_broker_transport::unix_socket::UnixStreamHostControlChannel::from_accepted(stream), + inner: litebox_broker_transport::unix_socket::UnixStreamHostControlChannel::from_accepted(control_stream), close_object_count: 0, }; - let termination = litebox_broker_host::serve_connection(&broker, &mut channel) - .expect("broker host failed"); + let mut notification_channel = + litebox_broker_transport::unix_socket::UnixStreamHostNotificationChannel::from_accepted(notification_stream); + let termination = litebox_broker_host::serve_connection_with_notifications( + &broker, + &mut channel, + &mut notification_channel, + ) + .expect("broker host failed"); assert_eq!( termination, litebox_broker_host::ConnectionTermination::PeerClosed @@ -341,7 +374,8 @@ fn spawn_test_broker( .expect("failed to report broker close-object count"); } })); - let _ = std::fs::remove_file(&server_socket_path); + let _ = std::fs::remove_file(&server_control_socket_path); + let _ = std::fs::remove_file(&server_notification_socket_path); let _ = done_tx.send(()); if let Err(panic) = result { std::panic::resume_unwind(panic); @@ -355,7 +389,8 @@ fn spawn_test_broker( thread: Some(broker_thread), done_rx, close_object_count_rx, - socket_path: cleanup_socket_path, + control_socket_path: cleanup_control_socket_path, + notification_socket_path: cleanup_notification_socket_path, } } @@ -428,9 +463,11 @@ impl fn test_runner_broker_integration_with_rewriter() { let true_path = run_which("true"); let target = common::compile("./tests/eventfd.c", "broker_eventfd_rewriter", false, false); - let socket_path = unique_test_socket_path("runner-broker"); + let control_socket_path = unique_test_socket_path("runner-broker-control"); + let notification_socket_path = unique_test_socket_path("runner-broker-notification"); let broker_thread = spawn_test_broker( - &socket_path, + &control_socket_path, + ¬ification_socket_path, litebox_broker_core::PolicyEngine::with_unauthenticated_rights( litebox_broker_core::PrincipalRights::all(), ), @@ -438,12 +475,12 @@ fn test_runner_broker_integration_with_rewriter() { ); Runner::new(&true_path, "broker_true_rewriter") - .broker_socket(&socket_path) + .broker_socket(&control_socket_path, ¬ification_socket_path) .run(); assert_eq!(broker_thread.next_close_object_count(), 0); Runner::new(&target, "broker_eventfd_rewriter") - .broker_socket(&socket_path) + .broker_socket(&control_socket_path, ¬ification_socket_path) .run(); // eventfd.c creates eight eventfd objects; each should release one broker object. assert_eq!(broker_thread.next_close_object_count(), 8); From 3b6cd954511569ea428e31a9e2490a588c5d004d Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 07:17:36 -0700 Subject: [PATCH 04/15] Rename paired broker serving API Rename the mandatory paired control/notification host serving entry point to serve_connection now that notification channels are no longer optional. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 12 ++++++------ litebox_broker_userland/src/main.rs | 10 ++++------ .../tests/notification_runtime.rs | 4 ++-- litebox_runner_linux_userland/tests/run.rs | 2 +- 4 files changed, 13 insertions(+), 15 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index 424b61dfea..8e7e0e27dc 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -34,7 +34,7 @@ pub use error::{BrokerHostError, Result}; /// The deployment must bind both channels to the same authenticated peer /// association. Active requests and responses remain on the control channel; /// broker-initiated readiness wakeups are sent on the notification channel. -pub fn serve_connection_with_notifications( +pub fn serve_connection( core: &BrokerCore, control_channel: &mut ControlChannel, notification_channel: &mut NotificationChannel, @@ -248,7 +248,7 @@ mod tests { let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), + serve_connection(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::PeerClosed ); assert_eq!( @@ -279,7 +279,7 @@ mod tests { let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), + serve_connection(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::PeerClosed ); assert_eq!( @@ -303,7 +303,7 @@ mod tests { let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), + serve_connection(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::ProtocolViolation ); assert_eq!( @@ -323,7 +323,7 @@ mod tests { let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_connection_with_notifications(broker, &mut channel, &mut notifications).unwrap(), + serve_connection(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::ProtocolViolation ); assert_eq!( @@ -348,7 +348,7 @@ mod tests { channel.send_error = true; let mut notifications = FakeHostNotificationChannel::default(); - match serve_connection_with_notifications(broker, &mut channel, &mut notifications) { + match serve_connection(broker, &mut channel, &mut notifications) { Err(BrokerHostError::Channel(())) => {} result => panic!("unexpected serve result: {result:?}"), } diff --git a/litebox_broker_userland/src/main.rs b/litebox_broker_userland/src/main.rs index 0c9b1bf070..0327e186cb 100644 --- a/litebox_broker_userland/src/main.rs +++ b/litebox_broker_userland/src/main.rs @@ -9,7 +9,7 @@ use std::process::Command; use clap::Parser; use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights}; -use litebox_broker_host::serve_connection_with_notifications; +use litebox_broker_host::serve_connection; use litebox_broker_transport::unix_socket::{ UnixStreamHostControlChannel, UnixStreamHostNotificationChannel, }; @@ -63,11 +63,9 @@ fn main() -> Result<(), Box> { UnixStreamHostControlChannel::from_accepted(control_stream); let mut notification_channel = UnixStreamHostNotificationChannel::from_accepted(notification_stream); - if let Err(error) = serve_connection_with_notifications( - &broker, - &mut control_channel, - &mut notification_channel, - ) { + if let Err(error) = + serve_connection(&broker, &mut control_channel, &mut notification_channel) + { eprintln!("failed to serve broker connection: {error}"); } }) diff --git a/litebox_broker_userland/tests/notification_runtime.rs b/litebox_broker_userland/tests/notification_runtime.rs index 97a15c21c8..8ad9ce30b6 100644 --- a/litebox_broker_userland/tests/notification_runtime.rs +++ b/litebox_broker_userland/tests/notification_runtime.rs @@ -5,7 +5,7 @@ use std::os::unix::net::UnixStream; use std::time::Duration; use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights}; -use litebox_broker_host::{ConnectionTermination, serve_connection_with_notifications}; +use litebox_broker_host::{ConnectionTermination, serve_connection}; use litebox_broker_local::{BrokerLocal, BrokerNotifications}; use litebox_broker_protocol::event::ReadinessState; use litebox_broker_protocol::message::{BrokerNotification, EventReadinessNotification}; @@ -29,7 +29,7 @@ fn host_sends_readiness_notifications_over_paired_userland_channel() { let host_thread = std::thread::spawn(move || { let mut control = UnixStreamHostControlChannel::from_accepted(host_control); let mut notification = UnixStreamHostNotificationChannel::from_accepted(host_notification); - serve_connection_with_notifications(&broker, &mut control, &mut notification) + serve_connection(&broker, &mut control, &mut notification) }); let mut local = diff --git a/litebox_runner_linux_userland/tests/run.rs b/litebox_runner_linux_userland/tests/run.rs index 653cd6bdf4..dfeca86655 100644 --- a/litebox_runner_linux_userland/tests/run.rs +++ b/litebox_runner_linux_userland/tests/run.rs @@ -359,7 +359,7 @@ fn spawn_test_broker( }; let mut notification_channel = litebox_broker_transport::unix_socket::UnixStreamHostNotificationChannel::from_accepted(notification_stream); - let termination = litebox_broker_host::serve_connection_with_notifications( + let termination = litebox_broker_host::serve_connection( &broker, &mut channel, &mut notification_channel, From a6ae87df8265b9308097f26055d5b63b67b0274a Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 07:19:50 -0700 Subject: [PATCH 05/15] Align broker host channel error bounds Use an explicit shared ChannelError generic for paired control and notification channels so their host serving bounds stay symmetric. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index 8e7e0e27dc..a4b7184e22 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -34,14 +34,14 @@ pub use error::{BrokerHostError, Result}; /// The deployment must bind both channels to the same authenticated peer /// association. Active requests and responses remain on the control channel; /// broker-initiated readiness wakeups are sent on the notification channel. -pub fn serve_connection( +pub fn serve_connection( core: &BrokerCore, control_channel: &mut ControlChannel, notification_channel: &mut NotificationChannel, -) -> Result +) -> Result where - ControlChannel: HostControlChannel, - NotificationChannel: HostNotificationChannel, + ControlChannel: HostControlChannel, + NotificationChannel: HostNotificationChannel, { let peer_credential = control_channel .peer_credential() @@ -90,14 +90,14 @@ where serve_request_loop(control_channel, notification_channel, &session) } -fn serve_request_loop( +fn serve_request_loop( control_channel: &mut ControlChannel, notification_channel: &mut NotificationChannel, session: &BrokerSession, -) -> Result +) -> Result where - ControlChannel: HostControlChannel, - NotificationChannel: HostNotificationChannel, + ControlChannel: HostControlChannel, + NotificationChannel: HostNotificationChannel, { loop { let request = match control_channel From f43f2f6273602b2a0149e681b478de42d1bfdfe8 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 07:26:42 -0700 Subject: [PATCH 06/15] Attach notifier to broker host connection Route readiness notifications through per-association host connection state instead of passing the notification channel directly through the request loop. This makes the notification channel the connection-owned broker notification path and keeps room for future notifications that are not direct request-loop side effects. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 75 +++++++++++++++++++++++++++++----- 1 file changed, 64 insertions(+), 11 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index a4b7184e22..e557b084ef 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -51,6 +51,10 @@ where _ => return Err(BrokerHostError::Broker(ErrorCode::PolicyDenied)), }; let session = core.create_session(caller_credential)?; + let mut connection = HostConnection { + session, + notification_channel, + }; loop { let request = match control_channel @@ -87,13 +91,12 @@ where } } - serve_request_loop(control_channel, notification_channel, &session) + serve_request_loop(control_channel, &mut connection) } fn serve_request_loop( control_channel: &mut ControlChannel, - notification_channel: &mut NotificationChannel, - session: &BrokerSession, + connection: &mut HostConnection<'_, NotificationChannel>, ) -> Result where ControlChannel: HostControlChannel, @@ -114,13 +117,13 @@ where HostReceive::PeerClosed => break, }; - let handled = handle_request(session, request); + let handled = handle_request(&connection.session, request); control_channel .send_response(&handled.response) .map_err(BrokerHostError::Channel)?; if let Some(notification) = handled.event_readiness_notification { - notification_channel - .send_notification(&BrokerNotification::EventReadiness(notification)) + connection + .notify_event_readiness(notification) .map_err(BrokerHostError::Channel)?; } } @@ -128,6 +131,25 @@ where Ok(ConnectionTermination::PeerClosed) } +/// Broker-host state for one authenticated control/notification association. +struct HostConnection<'a, NotificationChannel> { + session: BrokerSession, + notification_channel: &'a mut NotificationChannel, +} + +impl HostConnection<'_, NotificationChannel> +where + NotificationChannel: HostNotificationChannel, +{ + fn notify_event_readiness( + &mut self, + notification: EventReadinessNotification, + ) -> core::result::Result<(), NotificationChannel::Error> { + self.notification_channel + .send_notification(&BrokerNotification::EventReadiness(notification)) + } +} + fn handle_request(session: &BrokerSession, request: BrokerRequest) -> HandledRequest { match request { BrokerRequest::CloseObject(handle) => match session.close_object_reference(handle) { @@ -230,6 +252,7 @@ mod tests { serve_connection_rejects_handshake_request_after_negotiation(&broker); serve_connection_returns_channel_error_when_response_send_fails(&broker); serve_request_loop_sends_event_readiness_notifications(&broker); + host_connection_sends_event_readiness_notifications(&broker); active_request_closes_object_reference(&broker); } @@ -379,11 +402,16 @@ mod tests { ]), ); let mut notifications = FakeHostNotificationChannel::default(); - - assert_eq!( - serve_request_loop(&mut channel, &mut notifications, &session).unwrap(), - ConnectionTermination::PeerClosed - ); + { + let mut connection = HostConnection { + session, + notification_channel: &mut notifications, + }; + assert_eq!( + serve_request_loop(&mut channel, &mut connection).unwrap(), + ConnectionTermination::PeerClosed + ); + } assert_eq!( notifications.notifications, [ @@ -405,6 +433,31 @@ mod tests { ); } + fn host_connection_sends_event_readiness_notifications(broker: &BrokerCore) { + let session = broker + .create_session(CallerCredential::Unauthenticated) + .unwrap(); + let handle = litebox_broker_core::event::create(&session, 0).unwrap(); + let readiness = litebox_broker_protocol::event::ReadinessState { + read_ready: true, + write_ready: true, + }; + let notification = EventReadinessNotification { handle, readiness }; + let mut notifications = FakeHostNotificationChannel::default(); + { + let mut connection = HostConnection { + session, + notification_channel: &mut notifications, + }; + connection.notify_event_readiness(notification).unwrap(); + } + + assert_eq!( + notifications.notifications, + [BrokerNotification::EventReadiness(notification)] + ); + } + fn active_request_closes_object_reference(broker: &BrokerCore) { let session = broker .create_session(CallerCredential::Unauthenticated) From 88255344205c59936f85d711562fce96a57c1ceb Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 09:46:53 -0700 Subject: [PATCH 07/15] Emit readiness notifications from event handling Move readiness notification emission into the broker host event request handler so successful event state mutations trigger notifications at the event handling boundary rather than as request-loop postprocessing. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 130 +++++++++++++++++---------------- 1 file changed, 69 insertions(+), 61 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index e557b084ef..9da511f412 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -117,15 +117,10 @@ where HostReceive::PeerClosed => break, }; - let handled = handle_request(&connection.session, request); + let response = handle_request(connection, request).map_err(BrokerHostError::Channel)?; control_channel - .send_response(&handled.response) + .send_response(&response) .map_err(BrokerHostError::Channel)?; - if let Some(notification) = handled.event_readiness_notification { - connection - .notify_event_readiness(notification) - .map_err(BrokerHostError::Channel)?; - } } Ok(ConnectionTermination::PeerClosed) @@ -150,77 +145,85 @@ where } } -fn handle_request(session: &BrokerSession, request: BrokerRequest) -> HandledRequest { +fn handle_request( + connection: &mut HostConnection<'_, NotificationChannel>, + request: BrokerRequest, +) -> core::result::Result +where + NotificationChannel: HostNotificationChannel, +{ match request { - BrokerRequest::CloseObject(handle) => match session.close_object_reference(handle) { - Ok(()) => HandledRequest::response(BrokerResponse::ObjectClosed), - Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), - }, - BrokerRequest::Event(request) => handle_event_request(session, request), + BrokerRequest::CloseObject(handle) => { + match connection.session.close_object_reference(handle) { + Ok(()) => Ok(BrokerResponse::ObjectClosed), + Err(error) => Ok(BrokerResponse::Error(error.into())), + } + } + BrokerRequest::Event(request) => handle_event_request(connection, request), } } -fn handle_event_request(session: &BrokerSession, request: EventRequest) -> HandledRequest { +fn handle_event_request( + connection: &mut HostConnection<'_, NotificationChannel>, + request: EventRequest, +) -> core::result::Result +where + NotificationChannel: HostNotificationChannel, +{ match request { EventRequest::Create(request) => { - match litebox_broker_core::event::create(session, request.initial_count) { - Ok(handle) => HandledRequest::response(BrokerResponse::Event( - EventResponse::Create(CreateEventResponse { handle }), - )), - Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), + match litebox_broker_core::event::create(&connection.session, request.initial_count) { + Ok(handle) => Ok(BrokerResponse::Event(EventResponse::Create( + CreateEventResponse { handle }, + ))), + Err(error) => Ok(BrokerResponse::Error(error.into())), } } EventRequest::Wait(request) => { - match litebox_broker_core::event::wait(session, request.handle) { - Ok(readiness) => HandledRequest::response(BrokerResponse::Event( - EventResponse::Wait(WaitEventResponse { readiness }), - )), - Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), + match litebox_broker_core::event::wait(&connection.session, request.handle) { + Ok(readiness) => Ok(BrokerResponse::Event(EventResponse::Wait( + WaitEventResponse { readiness }, + ))), + Err(error) => Ok(BrokerResponse::Error(error.into())), } } EventRequest::Add(request) => { - match litebox_broker_core::event::add(session, request.handle, request.value) { - Ok(readiness) => HandledRequest { - response: BrokerResponse::Event(EventResponse::Add(AddEventResponse { - readiness, - })), - event_readiness_notification: Some(EventReadinessNotification { + match litebox_broker_core::event::add( + &connection.session, + request.handle, + request.value, + ) { + Ok(readiness) => { + connection.notify_event_readiness(EventReadinessNotification { handle: request.handle, readiness, - }), - }, - Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), + })?; + Ok(BrokerResponse::Event(EventResponse::Add( + AddEventResponse { readiness }, + ))) + } + Err(error) => Ok(BrokerResponse::Error(error.into())), } } EventRequest::Consume(request) => { - match litebox_broker_core::event::consume(session, request.handle, request.mode) { - Ok(consumption) => HandledRequest { - event_readiness_notification: Some(EventReadinessNotification { + match litebox_broker_core::event::consume( + &connection.session, + request.handle, + request.mode, + ) { + Ok(consumption) => { + connection.notify_event_readiness(EventReadinessNotification { handle: request.handle, readiness: consumption.readiness, - }), - response: BrokerResponse::Event(EventResponse::Consume(consumption)), - }, - Err(error) => HandledRequest::response(BrokerResponse::Error(error.into())), + })?; + Ok(BrokerResponse::Event(EventResponse::Consume(consumption))) + } + Err(error) => Ok(BrokerResponse::Error(error.into())), } } } } -struct HandledRequest { - response: BrokerResponse, - event_readiness_notification: Option, -} - -impl HandledRequest { - fn response(response: BrokerResponse) -> Self { - Self { - response, - event_readiness_notification: None, - } - } -} - /// Terminal outcome after processing one broker connection. #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[non_exhaustive] @@ -462,36 +465,41 @@ mod tests { let session = broker .create_session(CallerCredential::Unauthenticated) .unwrap(); + let mut notifications = FakeHostNotificationChannel::default(); + let mut connection = HostConnection { + session, + notification_channel: &mut notifications, + }; let response = handle_request( - &session, + &mut connection, BrokerRequest::Event(EventRequest::Create(CreateEventRequest { initial_count: 0, })), ) - .response; + .unwrap(); let BrokerResponse::Event(EventResponse::Create(response)) = response else { panic!("unexpected create response: {response:?}"); }; let handle = response.handle; assert_eq!( - handle_request(&session, BrokerRequest::CloseObject(handle)).response, + handle_request(&mut connection, BrokerRequest::CloseObject(handle)).unwrap(), BrokerResponse::ObjectClosed ); assert_eq!( handle_request( - &session, + &mut connection, BrokerRequest::Event(EventRequest::Wait(WaitEventRequest { handle })) ) - .response, + .unwrap(), BrokerResponse::Error(ErrorCode::UnknownObject) ); assert_eq!( handle_request( - &session, + &mut connection, BrokerRequest::CloseObject(ObjectHandle(handle.0 + 1)) ) - .response, + .unwrap(), BrokerResponse::Error(ErrorCode::UnknownObject) ); } From f85dc25a7fcd9305264149b027ca3e29056f1f6a Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 09:57:06 -0700 Subject: [PATCH 08/15] Avoid duplicate readiness notifications Do not emit readiness notifications for event mutations whose control responses already carry the resulting readiness state. Keep the paired notification channel mandatory so broker-originated readiness updates without a paired control response have a channel when those sources are added. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 212 +++++------------- .../tests/notification_runtime.rs | 21 +- .../tests/userland_broker.rs | 13 +- 3 files changed, 63 insertions(+), 183 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index 9da511f412..a338682079 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -20,8 +20,7 @@ use litebox_broker_protocol::channel::{ use litebox_broker_protocol::error::ErrorCode; use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse}; use litebox_broker_protocol::message::{ - BrokerHandshakeResponse, BrokerNotification, BrokerRequest, BrokerResponse, - EventReadinessNotification, EventRequest, EventResponse, + BrokerHandshakeResponse, BrokerRequest, BrokerResponse, EventRequest, EventResponse, }; mod error; @@ -34,10 +33,12 @@ pub use error::{BrokerHostError, Result}; /// The deployment must bind both channels to the same authenticated peer /// association. Active requests and responses remain on the control channel; /// broker-initiated readiness wakeups are sent on the notification channel. +/// Event mutations caused by control requests return readiness in their control +/// response and do not also emit a duplicate notification. pub fn serve_connection( core: &BrokerCore, control_channel: &mut ControlChannel, - notification_channel: &mut NotificationChannel, + _notification_channel: &mut NotificationChannel, ) -> Result where ControlChannel: HostControlChannel, @@ -51,10 +52,6 @@ where _ => return Err(BrokerHostError::Broker(ErrorCode::PolicyDenied)), }; let session = core.create_session(caller_credential)?; - let mut connection = HostConnection { - session, - notification_channel, - }; loop { let request = match control_channel @@ -91,16 +88,15 @@ where } } - serve_request_loop(control_channel, &mut connection) + serve_request_loop(control_channel, &session) } -fn serve_request_loop( +fn serve_request_loop( control_channel: &mut ControlChannel, - connection: &mut HostConnection<'_, NotificationChannel>, + session: &BrokerSession, ) -> Result where ControlChannel: HostControlChannel, - NotificationChannel: HostNotificationChannel, { loop { let request = match control_channel @@ -117,7 +113,7 @@ where HostReceive::PeerClosed => break, }; - let response = handle_request(connection, request).map_err(BrokerHostError::Channel)?; + let response = handle_request(session, request); control_channel .send_response(&response) .map_err(BrokerHostError::Channel)?; @@ -126,99 +122,46 @@ where Ok(ConnectionTermination::PeerClosed) } -/// Broker-host state for one authenticated control/notification association. -struct HostConnection<'a, NotificationChannel> { - session: BrokerSession, - notification_channel: &'a mut NotificationChannel, -} - -impl HostConnection<'_, NotificationChannel> -where - NotificationChannel: HostNotificationChannel, -{ - fn notify_event_readiness( - &mut self, - notification: EventReadinessNotification, - ) -> core::result::Result<(), NotificationChannel::Error> { - self.notification_channel - .send_notification(&BrokerNotification::EventReadiness(notification)) - } -} - -fn handle_request( - connection: &mut HostConnection<'_, NotificationChannel>, - request: BrokerRequest, -) -> core::result::Result -where - NotificationChannel: HostNotificationChannel, -{ +fn handle_request(session: &BrokerSession, request: BrokerRequest) -> BrokerResponse { match request { - BrokerRequest::CloseObject(handle) => { - match connection.session.close_object_reference(handle) { - Ok(()) => Ok(BrokerResponse::ObjectClosed), - Err(error) => Ok(BrokerResponse::Error(error.into())), - } - } - BrokerRequest::Event(request) => handle_event_request(connection, request), + BrokerRequest::CloseObject(handle) => match session.close_object_reference(handle) { + Ok(()) => BrokerResponse::ObjectClosed, + Err(error) => BrokerResponse::Error(error.into()), + }, + BrokerRequest::Event(request) => handle_event_request(session, request), } } -fn handle_event_request( - connection: &mut HostConnection<'_, NotificationChannel>, - request: EventRequest, -) -> core::result::Result -where - NotificationChannel: HostNotificationChannel, -{ +fn handle_event_request(session: &BrokerSession, request: EventRequest) -> BrokerResponse { match request { EventRequest::Create(request) => { - match litebox_broker_core::event::create(&connection.session, request.initial_count) { - Ok(handle) => Ok(BrokerResponse::Event(EventResponse::Create( - CreateEventResponse { handle }, - ))), - Err(error) => Ok(BrokerResponse::Error(error.into())), + match litebox_broker_core::event::create(session, request.initial_count) { + Ok(handle) => { + BrokerResponse::Event(EventResponse::Create(CreateEventResponse { handle })) + } + Err(error) => BrokerResponse::Error(error.into()), } } EventRequest::Wait(request) => { - match litebox_broker_core::event::wait(&connection.session, request.handle) { - Ok(readiness) => Ok(BrokerResponse::Event(EventResponse::Wait( - WaitEventResponse { readiness }, - ))), - Err(error) => Ok(BrokerResponse::Error(error.into())), + match litebox_broker_core::event::wait(session, request.handle) { + Ok(readiness) => { + BrokerResponse::Event(EventResponse::Wait(WaitEventResponse { readiness })) + } + Err(error) => BrokerResponse::Error(error.into()), } } EventRequest::Add(request) => { - match litebox_broker_core::event::add( - &connection.session, - request.handle, - request.value, - ) { + match litebox_broker_core::event::add(session, request.handle, request.value) { Ok(readiness) => { - connection.notify_event_readiness(EventReadinessNotification { - handle: request.handle, - readiness, - })?; - Ok(BrokerResponse::Event(EventResponse::Add( - AddEventResponse { readiness }, - ))) + BrokerResponse::Event(EventResponse::Add(AddEventResponse { readiness })) } - Err(error) => Ok(BrokerResponse::Error(error.into())), + Err(error) => BrokerResponse::Error(error.into()), } } EventRequest::Consume(request) => { - match litebox_broker_core::event::consume( - &connection.session, - request.handle, - request.mode, - ) { - Ok(consumption) => { - connection.notify_event_readiness(EventReadinessNotification { - handle: request.handle, - readiness: consumption.readiness, - })?; - Ok(BrokerResponse::Event(EventResponse::Consume(consumption))) - } - Err(error) => Ok(BrokerResponse::Error(error.into())), + match litebox_broker_core::event::consume(session, request.handle, request.mode) { + Ok(consumption) => BrokerResponse::Event(EventResponse::Consume(consumption)), + Err(error) => BrokerResponse::Error(error.into()), } } } @@ -239,7 +182,7 @@ mod tests { use super::*; use litebox_broker_core::{PolicyEngine, PrincipalRights}; use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest}; - use litebox_broker_protocol::message::BrokerHandshakeRequest; + use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification}; use litebox_broker_protocol::{ObjectHandle, ProtocolVersion}; #[test] @@ -254,8 +197,7 @@ mod tests { serve_connection_rejects_active_request_before_negotiation(&broker); serve_connection_rejects_handshake_request_after_negotiation(&broker); serve_connection_returns_channel_error_when_response_send_fails(&broker); - serve_request_loop_sends_event_readiness_notifications(&broker); - host_connection_sends_event_readiness_notifications(&broker); + serve_request_loop_returns_event_readiness_in_control_responses(&broker); active_request_closes_object_reference(&broker); } @@ -381,7 +323,7 @@ mod tests { assert!(channel.handshake_responses.is_empty()); } - fn serve_request_loop_sends_event_readiness_notifications(broker: &BrokerCore) { + fn serve_request_loop_returns_event_readiness_in_control_responses(broker: &BrokerCore) { let session = broker .create_session(CallerCredential::Unauthenticated) .unwrap(); @@ -404,102 +346,64 @@ mod tests { Ok(HostReceive::PeerClosed), ]), ); - let mut notifications = FakeHostNotificationChannel::default(); - { - let mut connection = HostConnection { - session, - notification_channel: &mut notifications, - }; - assert_eq!( - serve_request_loop(&mut channel, &mut connection).unwrap(), - ConnectionTermination::PeerClosed - ); - } + assert_eq!( - notifications.notifications, + serve_request_loop(&mut channel, &session).unwrap(), + ConnectionTermination::PeerClosed + ); + assert_eq!( + channel.responses, [ - BrokerNotification::EventReadiness(EventReadinessNotification { - handle, + BrokerResponse::Event(EventResponse::Add(AddEventResponse { readiness: litebox_broker_protocol::event::ReadinessState { read_ready: true, write_ready: true, }, - }), - BrokerNotification::EventReadiness(EventReadinessNotification { - handle, - readiness: litebox_broker_protocol::event::ReadinessState { - read_ready: false, - write_ready: true, - }, - }), + })), + BrokerResponse::Event(EventResponse::Consume( + litebox_broker_protocol::event::ConsumeEventResponse { + value: 1, + readiness: litebox_broker_protocol::event::ReadinessState { + read_ready: false, + write_ready: true, + }, + } + )), ] ); } - fn host_connection_sends_event_readiness_notifications(broker: &BrokerCore) { - let session = broker - .create_session(CallerCredential::Unauthenticated) - .unwrap(); - let handle = litebox_broker_core::event::create(&session, 0).unwrap(); - let readiness = litebox_broker_protocol::event::ReadinessState { - read_ready: true, - write_ready: true, - }; - let notification = EventReadinessNotification { handle, readiness }; - let mut notifications = FakeHostNotificationChannel::default(); - { - let mut connection = HostConnection { - session, - notification_channel: &mut notifications, - }; - connection.notify_event_readiness(notification).unwrap(); - } - - assert_eq!( - notifications.notifications, - [BrokerNotification::EventReadiness(notification)] - ); - } - fn active_request_closes_object_reference(broker: &BrokerCore) { let session = broker .create_session(CallerCredential::Unauthenticated) .unwrap(); - let mut notifications = FakeHostNotificationChannel::default(); - let mut connection = HostConnection { - session, - notification_channel: &mut notifications, - }; let response = handle_request( - &mut connection, + &session, BrokerRequest::Event(EventRequest::Create(CreateEventRequest { initial_count: 0, })), - ) - .unwrap(); + ); let BrokerResponse::Event(EventResponse::Create(response)) = response else { panic!("unexpected create response: {response:?}"); }; let handle = response.handle; assert_eq!( - handle_request(&mut connection, BrokerRequest::CloseObject(handle)).unwrap(), + handle_request(&session, BrokerRequest::CloseObject(handle)), BrokerResponse::ObjectClosed ); assert_eq!( handle_request( - &mut connection, + &session, BrokerRequest::Event(EventRequest::Wait(WaitEventRequest { handle })) - ) - .unwrap(), + ), BrokerResponse::Error(ErrorCode::UnknownObject) ); assert_eq!( handle_request( - &mut connection, + &session, BrokerRequest::CloseObject(ObjectHandle(handle.0 + 1)) - ) - .unwrap(), + ), BrokerResponse::Error(ErrorCode::UnknownObject) ); } diff --git a/litebox_broker_userland/tests/notification_runtime.rs b/litebox_broker_userland/tests/notification_runtime.rs index 8ad9ce30b6..9fd6fb3aa3 100644 --- a/litebox_broker_userland/tests/notification_runtime.rs +++ b/litebox_broker_userland/tests/notification_runtime.rs @@ -2,29 +2,23 @@ // Licensed under the MIT license. use std::os::unix::net::UnixStream; -use std::time::Duration; use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights}; use litebox_broker_host::{ConnectionTermination, serve_connection}; -use litebox_broker_local::{BrokerLocal, BrokerNotifications}; +use litebox_broker_local::BrokerLocal; use litebox_broker_protocol::event::ReadinessState; -use litebox_broker_protocol::message::{BrokerNotification, EventReadinessNotification}; use litebox_broker_transport::unix_socket::{ UnixStreamHostControlChannel, UnixStreamHostNotificationChannel, UnixStreamLocalControlChannel, - UnixStreamLocalNotificationChannel, }; #[test] -fn host_sends_readiness_notifications_over_paired_userland_channel() { +fn host_serves_control_requests_over_paired_userland_channels() { let broker = BrokerCore::new(PolicyEngine::with_unauthenticated_rights( PrincipalRights::all(), )) .unwrap(); let (local_control, host_control) = UnixStream::pair().unwrap(); - let (local_notification, host_notification) = UnixStream::pair().unwrap(); - local_notification - .set_read_timeout(Some(Duration::from_secs(5))) - .unwrap(); + let (_local_notification, host_notification) = UnixStream::pair().unwrap(); let host_thread = std::thread::spawn(move || { let mut control = UnixStreamHostControlChannel::from_accepted(host_control); @@ -35,9 +29,6 @@ fn host_sends_readiness_notifications_over_paired_userland_channel() { let mut local = BrokerLocal::negotiate(UnixStreamLocalControlChannel::from_connected(local_control)) .unwrap(); - let mut notifications = BrokerNotifications::new( - UnixStreamLocalNotificationChannel::from_connected(local_notification), - ); let handle = local.create_event_with_count(0).unwrap(); let readiness = ReadinessState { @@ -45,12 +36,6 @@ fn host_sends_readiness_notifications_over_paired_userland_channel() { write_ready: true, }; assert_eq!(local.add_event(handle, 1).unwrap(), readiness); - assert_eq!( - notifications.recv_notification().unwrap(), - Some(BrokerNotification::EventReadiness( - EventReadinessNotification { handle, readiness } - )) - ); drop(local); assert_eq!( diff --git a/litebox_broker_userland/tests/userland_broker.rs b/litebox_broker_userland/tests/userland_broker.rs index 3f98489a2b..cab3fccbf7 100644 --- a/litebox_broker_userland/tests/userland_broker.rs +++ b/litebox_broker_userland/tests/userland_broker.rs @@ -8,9 +8,8 @@ use std::path::Path; use std::process::{Child, Command}; use std::time::{Duration, Instant}; -use litebox_broker_local::{BrokerLocal, BrokerNotifications}; +use litebox_broker_local::BrokerLocal; use litebox_broker_protocol::event::ReadinessState; -use litebox_broker_protocol::message::{BrokerNotification, EventReadinessNotification}; use litebox_broker_transport::unix_socket::{ UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel, }; @@ -80,10 +79,9 @@ fn run_fake_runner(args: &[OsString]) { let control_socket_path = args.get(2).unwrap(); let notification_socket_path = args.get(4).unwrap(); let control_channel = connect_control_with_retry(Path::new(control_socket_path)).unwrap(); - let notification_channel = + let _notification_channel = connect_notification_with_retry(Path::new(notification_socket_path)).unwrap(); let mut local = BrokerLocal::negotiate(control_channel).unwrap(); - let mut notifications = BrokerNotifications::new(notification_channel); let handle = local.create_event_with_count(0).unwrap(); assert_eq!( @@ -99,12 +97,6 @@ fn run_fake_runner(args: &[OsString]) { write_ready: true, }; assert_eq!(local.add_event(handle, 1).unwrap(), readiness); - assert_eq!( - notifications.recv_notification().unwrap(), - Some(BrokerNotification::EventReadiness( - EventReadinessNotification { handle, readiness } - )) - ); assert_eq!( local.wait_event(handle).unwrap(), @@ -113,7 +105,6 @@ fn run_fake_runner(args: &[OsString]) { write_ready: true, } ); - drop(notifications); drop(local); // SAFETY: `getppid` takes no pointer arguments and has no Rust-side aliasing requirements. From c37ef5a2c0a4eb9eb003c997c9389f724d5c8b60 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:00:12 -0700 Subject: [PATCH 09/15] Exercise readiness responses through serve_connection Update the broker host readiness-response test to drive the public paired serve_connection API. The fake channel now queues Add and Consume requests after observing the broker-assigned create handle. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 63 ++++++++++++++++++++-------------- 1 file changed, 38 insertions(+), 25 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index a338682079..fbbff90244 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -181,7 +181,10 @@ pub enum ConnectionTermination { mod tests { use super::*; use litebox_broker_core::{PolicyEngine, PrincipalRights}; - use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest}; + use litebox_broker_protocol::event::{ + AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode, + WaitEventRequest, + }; use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification}; use litebox_broker_protocol::{ObjectHandle, ProtocolVersion}; @@ -197,7 +200,7 @@ mod tests { serve_connection_rejects_active_request_before_negotiation(&broker); serve_connection_rejects_handshake_request_after_negotiation(&broker); serve_connection_returns_channel_error_when_response_send_fails(&broker); - serve_request_loop_returns_event_readiness_in_control_responses(&broker); + serve_connection_returns_event_readiness_in_control_responses(&broker); active_request_closes_object_reference(&broker); } @@ -323,36 +326,25 @@ mod tests { assert!(channel.handshake_responses.is_empty()); } - fn serve_request_loop_returns_event_readiness_in_control_responses(broker: &BrokerCore) { - let session = broker - .create_session(CallerCredential::Unauthenticated) - .unwrap(); - let handle = litebox_broker_core::event::create(&session, 0).unwrap(); + fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) { let mut channel = FakeHostControlChannel::new( - std::vec::Vec::new(), - std::vec::Vec::from([ - Ok(HostReceive::Message(BrokerRequest::Event( - EventRequest::Add(litebox_broker_protocol::event::AddEventRequest { - handle, - value: 1, - }), - ))), - Ok(HostReceive::Message(BrokerRequest::Event( - EventRequest::Consume(litebox_broker_protocol::event::ConsumeEventRequest { - handle, - mode: litebox_broker_protocol::event::EventConsumeMode::One, - }), - ))), - Ok(HostReceive::PeerClosed), - ]), + std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest { + protocol_version: BROKER_PROTOCOL_VERSION, + }))]), + std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event( + EventRequest::Create(CreateEventRequest { initial_count: 0 }), + )))]), ); + channel.enqueue_readiness_requests_after_create = true; + let mut notifications = FakeHostNotificationChannel::default(); assert_eq!( - serve_request_loop(&mut channel, &session).unwrap(), + serve_connection(broker, &mut channel, &mut notifications).unwrap(), ConnectionTermination::PeerClosed ); + assert!(notifications.notifications.is_empty()); assert_eq!( - channel.responses, + &channel.responses[1..], [ BrokerResponse::Event(EventResponse::Add(AddEventResponse { readiness: litebox_broker_protocol::event::ReadinessState { @@ -414,6 +406,7 @@ mod tests { requests: std::vec::Vec, ()>>, handshake_responses: std::vec::Vec, responses: std::vec::Vec, + enqueue_readiness_requests_after_create: bool, send_error: bool, } @@ -429,6 +422,7 @@ mod tests { requests, handshake_responses: std::vec::Vec::new(), responses: std::vec::Vec::new(), + enqueue_readiness_requests_after_create: false, send_error: false, } } @@ -479,6 +473,25 @@ mod tests { if self.send_error { return Err(()); } + if self.enqueue_readiness_requests_after_create + && let BrokerResponse::Event(EventResponse::Create(response)) = response + { + self.requests + .push(Ok(HostReceive::Message(BrokerRequest::Event( + EventRequest::Add(AddEventRequest { + handle: response.handle, + value: 1, + }), + )))); + self.requests + .push(Ok(HostReceive::Message(BrokerRequest::Event( + EventRequest::Consume(ConsumeEventRequest { + handle: response.handle, + mode: EventConsumeMode::One, + }), + )))); + self.requests.push(Ok(HostReceive::PeerClosed)); + } self.responses.push(response.clone()); Ok(()) } From 8c1e075b5f9341773810c05b161c834e2420afa9 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:01:18 -0700 Subject: [PATCH 10/15] Inline broker host request loop Fold the private serve_request_loop helper into serve_connection now that tests exercise the public paired serving entry point directly. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_host/src/lib.rs | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index fbbff90244..905745bb5c 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -88,16 +88,6 @@ where } } - serve_request_loop(control_channel, &session) -} - -fn serve_request_loop( - control_channel: &mut ControlChannel, - session: &BrokerSession, -) -> Result -where - ControlChannel: HostControlChannel, -{ loop { let request = match control_channel .recv_request() @@ -113,7 +103,7 @@ where HostReceive::PeerClosed => break, }; - let response = handle_request(session, request); + let response = handle_request(&session, request); control_channel .send_response(&response) .map_err(BrokerHostError::Channel)?; From 3eb6ec55f304d1b46d158b664b05786b91a3e0c6 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:13:19 -0700 Subject: [PATCH 11/15] Rename broker control socket argument Rename the hidden runner control-channel argument from --broker-socket to --broker-control-socket so it matches the paired --broker-notification-socket argument. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_broker_userland/src/main.rs | 2 +- litebox_broker_userland/tests/userland_broker.rs | 4 ++-- litebox_runner_linux_userland/src/lib.rs | 6 +++--- litebox_runner_linux_userland/tests/run.rs | 10 ++++++---- 4 files changed, 12 insertions(+), 10 deletions(-) diff --git a/litebox_broker_userland/src/main.rs b/litebox_broker_userland/src/main.rs index 0327e186cb..5363b308bf 100644 --- a/litebox_broker_userland/src/main.rs +++ b/litebox_broker_userland/src/main.rs @@ -40,7 +40,7 @@ fn main() -> Result<(), Box> { let mut runner_command = Command::new(&args.runner); runner_command .arg("--unstable") - .arg("--broker-socket") + .arg("--broker-control-socket") .arg(&control_socket_path) .arg("--broker-notification-socket") .arg(¬ification_socket_path) diff --git a/litebox_broker_userland/tests/userland_broker.rs b/litebox_broker_userland/tests/userland_broker.rs index cab3fccbf7..ee126be51b 100644 --- a/litebox_broker_userland/tests/userland_broker.rs +++ b/litebox_broker_userland/tests/userland_broker.rs @@ -32,7 +32,7 @@ fn run_parent_test() { // This custom-harness integration test uses its own executable as the broker's // runner. Cargo starts this executable without broker args, so it runs the // parent path here. The broker then starts the same executable with the real - // runner argv (`--unstable --broker-socket + // runner argv (`--unstable --broker-control-socket // --broker-notification-socket `), which runs `run_fake_runner`. After // the fake runner finishes its broker requests, it terminates the broker // parent process; this lets the test exercise the long-running broker @@ -64,7 +64,7 @@ fn run_fake_runner(args: &[OsString]) { ); assert_eq!( args.get(1).map(OsString::as_os_str), - Some(OsStr::new("--broker-socket")) + Some(OsStr::new("--broker-control-socket")) ); assert_eq!( args.get(3).map(OsString::as_os_str), diff --git a/litebox_runner_linux_userland/src/lib.rs b/litebox_runner_linux_userland/src/lib.rs index e01997e874..c6ca262d2e 100644 --- a/litebox_runner_linux_userland/src/lib.rs +++ b/litebox_runner_linux_userland/src/lib.rs @@ -81,14 +81,14 @@ pub struct CliArgs { pub program_from_tar: bool, /// Broker-supplied Unix socket path for the local control channel. #[arg( - long = "broker-socket", + long = "broker-control-socket", value_name = "PATH", value_hint = clap::ValueHint::FilePath, hide = true, requires = "unstable", help_heading = "Unstable Options" )] - pub broker_socket: Option, + pub broker_control_socket: Option, /// Broker-supplied Unix socket path for the local notification channel. #[arg( long = "broker-notification-socket", @@ -224,7 +224,7 @@ pub fn run(cli_args: CliArgs) -> Result<()> { litebox_platform_multiplex::set_platform(platform); let broker_connection = broker::connect( - cli_args.broker_socket.as_deref(), + cli_args.broker_control_socket.as_deref(), cli_args.broker_notification_socket.as_deref(), )?; diff --git a/litebox_runner_linux_userland/tests/run.rs b/litebox_runner_linux_userland/tests/run.rs index dfeca86655..310636b955 100644 --- a/litebox_runner_linux_userland/tests/run.rs +++ b/litebox_runner_linux_userland/tests/run.rs @@ -123,12 +123,14 @@ impl Runner { } #[cfg(all(target_arch = "x86_64", target_os = "linux"))] - fn broker_socket( + fn broker_sockets( &mut self, control_socket_path: &Path, notification_socket_path: &Path, ) -> &mut Self { - self.command.arg("--broker-socket").arg(control_socket_path); + self.command + .arg("--broker-control-socket") + .arg(control_socket_path); self.command .arg("--broker-notification-socket") .arg(notification_socket_path); @@ -475,12 +477,12 @@ fn test_runner_broker_integration_with_rewriter() { ); Runner::new(&true_path, "broker_true_rewriter") - .broker_socket(&control_socket_path, ¬ification_socket_path) + .broker_sockets(&control_socket_path, ¬ification_socket_path) .run(); assert_eq!(broker_thread.next_close_object_count(), 0); Runner::new(&target, "broker_eventfd_rewriter") - .broker_socket(&control_socket_path, ¬ification_socket_path) + .broker_sockets(&control_socket_path, ¬ification_socket_path) .run(); // eventfd.c creates eight eventfd objects; each should release one broker object. assert_eq!(broker_thread.next_close_object_count(), 8); From 7f9b0743b06e10becc2754dd993e693ad7d70439 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:16:11 -0700 Subject: [PATCH 12/15] Clarify notification receiver thread ownership Rename the retained notification receiver thread handle to notification_receiver_thread and document that it is intentionally kept alive with the broker connection. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_runner_linux_userland/src/broker.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs index 097bda62b8..0078ea9ae8 100644 --- a/litebox_runner_linux_userland/src/broker.rs +++ b/litebox_runner_linux_userland/src/broker.rs @@ -19,7 +19,11 @@ type Local = BrokerLocal; pub(crate) struct BrokerConnection { local: Local, - _notification_thread: JoinHandle<()>, + #[expect( + dead_code, + reason = "keeps the notification receiver thread alive while the broker connection is installed" + )] + notification_receiver_thread: JoinHandle<()>, } pub(crate) fn connect( @@ -86,7 +90,7 @@ fn connect_to_endpoint( .context("failed to start broker notification receiver")?; Ok(BrokerConnection { local, - _notification_thread: notification_thread, + notification_receiver_thread: notification_thread, }) } From 9a185f3f1a56fd74b6433631a1a72ca3963f97d7 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:18:23 -0700 Subject: [PATCH 13/15] Simplify broker socket option handling Replace the tuple match in runner broker connection setup with straightforward early returns while preserving errors for mismatched control and notification socket arguments. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_runner_linux_userland/src/broker.rs | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs index 0078ea9ae8..4308c719aa 100644 --- a/litebox_runner_linux_userland/src/broker.rs +++ b/litebox_runner_linux_userland/src/broker.rs @@ -30,18 +30,16 @@ pub(crate) fn connect( control_socket_path: Option<&Path>, notification_socket_path: Option<&Path>, ) -> Result> { - match (control_socket_path, notification_socket_path) { - (Some(control_path), Some(notification_path)) => { - connect_to_endpoint(control_path, notification_path).map(Some) + let Some(control_socket_path) = control_socket_path else { + if notification_socket_path.is_some() { + bail!("broker control socket is required with broker notification socket"); } - (None, None) => Ok(None), - (Some(_), None) => { - bail!("broker notification socket is required with broker control socket") - } - (None, Some(_)) => { - bail!("broker control socket is required with broker notification socket") - } - } + return Ok(None); + }; + let Some(notification_socket_path) = notification_socket_path else { + bail!("broker notification socket is required with broker control socket"); + }; + connect_to_endpoint(control_socket_path, notification_socket_path).map(Some) } impl BrokerConnection { From 5597b375940d375f9d7ec070b5a78bb64fdb5e68 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:20:44 -0700 Subject: [PATCH 14/15] Require broker socket paths in connect Make runner broker::connect take concrete control and notification socket paths. The CLI layer now handles optional broker enablement and validates that hidden broker socket arguments are provided as a pair. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_runner_linux_userland/src/broker.rs | 30 +++++---------------- litebox_runner_linux_userland/src/lib.rs | 20 +++++++++++--- 2 files changed, 23 insertions(+), 27 deletions(-) diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs index 4308c719aa..139a7ae6c2 100644 --- a/litebox_runner_linux_userland/src/broker.rs +++ b/litebox_runner_linux_userland/src/broker.rs @@ -7,7 +7,7 @@ use std::{ time::{Duration, Instant}, }; -use anyhow::{Context as _, Result, bail}; +use anyhow::{Context as _, Result}; use litebox_broker_local::{BrokerLocal, BrokerNotifications}; use litebox_broker_transport::unix_socket::{ UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel, @@ -27,28 +27,6 @@ pub(crate) struct BrokerConnection { } pub(crate) fn connect( - control_socket_path: Option<&Path>, - notification_socket_path: Option<&Path>, -) -> Result> { - let Some(control_socket_path) = control_socket_path else { - if notification_socket_path.is_some() { - bail!("broker control socket is required with broker notification socket"); - } - return Ok(None); - }; - let Some(notification_socket_path) = notification_socket_path else { - bail!("broker notification socket is required with broker control socket"); - }; - connect_to_endpoint(control_socket_path, notification_socket_path).map(Some) -} - -impl BrokerConnection { - pub(crate) fn into_local(self) -> Local { - self.local - } -} - -fn connect_to_endpoint( control_socket_path: &Path, notification_socket_path: &Path, ) -> Result { @@ -92,6 +70,12 @@ fn connect_to_endpoint( }) } +impl BrokerConnection { + pub(crate) fn into_local(self) -> Local { + self.local + } +} + fn connect_control_with_retry( socket_path: &Path, setup_deadline: Instant, diff --git a/litebox_runner_linux_userland/src/lib.rs b/litebox_runner_linux_userland/src/lib.rs index c6ca262d2e..76c4e7b5e8 100644 --- a/litebox_runner_linux_userland/src/lib.rs +++ b/litebox_runner_linux_userland/src/lib.rs @@ -85,7 +85,7 @@ pub struct CliArgs { value_name = "PATH", value_hint = clap::ValueHint::FilePath, hide = true, - requires = "unstable", + requires_all = ["unstable", "broker_notification_socket"], help_heading = "Unstable Options" )] pub broker_control_socket: Option, @@ -95,7 +95,7 @@ pub struct CliArgs { value_name = "PATH", value_hint = clap::ValueHint::FilePath, hide = true, - requires = "unstable", + requires_all = ["unstable", "broker_control_socket"], help_heading = "Unstable Options" )] pub broker_notification_socket: Option, @@ -223,10 +223,22 @@ pub fn run(cli_args: CliArgs) -> Result<()> { } litebox_platform_multiplex::set_platform(platform); - let broker_connection = broker::connect( + let broker_connection = match ( cli_args.broker_control_socket.as_deref(), cli_args.broker_notification_socket.as_deref(), - )?; + ) { + (Some(control_socket_path), Some(notification_socket_path)) => Some(broker::connect( + control_socket_path, + notification_socket_path, + )?), + (None, None) => None, + (Some(_), None) => { + anyhow::bail!("broker notification socket is required with broker control socket") + } + (None, Some(_)) => { + anyhow::bail!("broker control socket is required with broker notification socket") + } + }; let shim_builder = if let Some(broker_connection) = broker_connection { litebox_shim_linux::LinuxShimBuilder::new_with_litebox( From 73f09bc09ec768cf2a05a2e01588dcda65df1509 Mon Sep 17 00:00:00 2001 From: Weidong Cui Date: Wed, 1 Jul 2026 10:24:26 -0700 Subject: [PATCH 15/15] Merge broker socket retry helpers Replace separate control and notification socket retry loops with one connect_with_retry helper parameterized by the channel-specific connect operation and timeout message. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- litebox_runner_linux_userland/src/broker.rs | 71 +++++++++------------ 1 file changed, 30 insertions(+), 41 deletions(-) diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs index 139a7ae6c2..7d36e96be9 100644 --- a/litebox_runner_linux_userland/src/broker.rs +++ b/litebox_runner_linux_userland/src/broker.rs @@ -31,22 +31,30 @@ pub(crate) fn connect( notification_socket_path: &Path, ) -> Result { let setup_deadline = Instant::now() + SETUP_TIMEOUT; - let control_channel = connect_control_with_retry(control_socket_path, setup_deadline) - .with_context(|| { - format!( - "failed to connect to broker at {}", - control_socket_path.display() - ) - })?; - let notification_channel = - connect_notification_with_retry(notification_socket_path, setup_deadline).with_context( - || { - format!( - "failed to connect to broker notifications at {}", - notification_socket_path.display() - ) - }, - )?; + let control_channel = connect_with_retry( + control_socket_path, + setup_deadline, + "timed out connecting to broker", + |path, deadline| UnixStreamLocalControlChannel::connect_with_setup_deadline(path, deadline), + ) + .with_context(|| { + format!( + "failed to connect to broker at {}", + control_socket_path.display() + ) + })?; + let notification_channel = connect_with_retry( + notification_socket_path, + setup_deadline, + "timed out connecting to broker notifications", + |path, _deadline| UnixStreamLocalNotificationChannel::connect(path), + ) + .with_context(|| { + format!( + "failed to connect to broker notifications at {}", + notification_socket_path.display() + ) + })?; let local = BrokerLocal::negotiate(control_channel).context("broker negotiation failed")?; let mut notifications = BrokerNotifications::new(notification_channel); let notification_thread = std::thread::Builder::new() @@ -76,37 +84,18 @@ impl BrokerConnection { } } -fn connect_control_with_retry( +fn connect_with_retry( socket_path: &Path, setup_deadline: Instant, -) -> Result { + timeout_message: &'static str, + mut connect: impl FnMut(&Path, Instant) -> std::io::Result, +) -> Result { loop { - match UnixStreamLocalControlChannel::connect_with_setup_deadline( - socket_path, - setup_deadline, - ) { + match connect(socket_path, setup_deadline) { Ok(channel) => return Ok(channel), Err(error) => { if Instant::now() >= setup_deadline { - return Err(error).context("timed out connecting to broker"); - } - } - } - let remaining = setup_deadline.saturating_duration_since(Instant::now()); - std::thread::sleep(RETRY_DELAY.min(remaining)); - } -} - -fn connect_notification_with_retry( - socket_path: &Path, - setup_deadline: Instant, -) -> Result { - loop { - match UnixStreamLocalNotificationChannel::connect(socket_path) { - Ok(channel) => return Ok(channel), - Err(error) => { - if Instant::now() >= setup_deadline { - return Err(error).context("timed out connecting to broker notifications"); + return Err(error).context(timeout_message); } } }