running_process/broker/server/
serve.rs

1//! Broker serve-mode wiring for registered and launch-backed backends.
2//!
3//! Phase 4 grows the long-lived daemon incrementally. This module connects the
4//! existing service-definition loader, broker instance routing, backend
5//! registry, backend launch coordination, and framed local-socket accept loop.
6//! Tests can still request bounded runs while the CLI defaults to accepting
7//! until process exit.
8
9use std::num::NonZeroUsize;
10use std::path::PathBuf;
11use std::sync::Mutex;
12use std::time::{Duration, Instant};
13
14use crate::broker::backend_handle::{BackendHandle, BackendHandleError, DaemonProcess};
15use crate::broker::backend_lifecycle::identity::IdentityError;
16use crate::broker::lifecycle::sid::SidError;
17use crate::broker::protocol::{Endpoint, ServiceDefinition};
18
19use super::admin::AdminSnapshot;
20use super::backend_launcher::{BackendLauncher, CommandBackendLauncher};
21use super::backend_registry::BackendRegistry;
22use super::combined_service_def_loader::CombinedServiceDefinitionLoader;
23use super::connection::{BrokerConnectionError, PeerCredentialPolicy};
24use super::control_socket::{
25    serve_control_socket_connections_with_limit_policy_post_hello_opaque,
26    serve_launch_control_socket_connections_concurrently, ControlSocketConnectionLimit,
27    ControlSocketError,
28};
29use super::fd_pressure::FdPressureGuard;
30use super::handoff_serve::{try_complete_negotiated_handoff_opaque, ServeHandoffContext};
31use super::hello_handler::{HelloHandler, HelloHandlerError};
32use super::hello_router::HelloRouter;
33use super::instance::{BrokerInstanceError, BrokerInstanceKey};
34use super::service_def_loader::{service_definition_dir, ServiceDefinitionError};
35use super::spawn_coordinator::SpawnCoordinator;
36use super::version_allow_list::{check_version_allowed, VersionPolicyBlock};
37
38/// Configuration for a bounded broker serve-mode run.
39#[derive(Clone, Debug)]
40pub struct BrokerServeConfig {
41    /// Local socket path or Windows pipe name to bind.
42    pub socket_path: String,
43    /// Service definition to load.
44    pub service_name: String,
45    /// Backend version to register for Hello negotiation.
46    pub service_version: String,
47    /// Direct backend endpoint returned to negotiated clients.
48    pub backend_endpoint: String,
49    /// Directory containing `<service>.servicedef` protobuf files.
50    pub service_definition_dir: PathBuf,
51    /// Optional number of control-socket connections to accept before returning.
52    pub max_connections: Option<NonZeroUsize>,
53    /// Optional backend handoff endpoint enabling the Phase 6 handle-passing
54    /// optimization (#387). `None` (the default) disables handoff entirely:
55    /// negotiated clients always reconnect through `backend_endpoint`. This
56    /// matches the opt-in Phase 6 gate in `docs/v1-rollout-policy.md`.
57    pub handoff_endpoint: Option<String>,
58    /// Optional deadline for the startup identity probe this serve run
59    /// performs against its own backend endpoint.
60    ///
61    /// `None` (the default) uses
62    /// [`probe::DEFAULT_ENDPOINT_PROBE_TIMEOUT`](crate::broker::backend_lifecycle::probe::DEFAULT_ENDPOINT_PROBE_TIMEOUT),
63    /// which is the right budget for a backend running at normal speed. Set it
64    /// when the backend is known to be slower for a reason unrelated to health
65    /// -- coverage instrumentation is the case this exists for (#1114).
66    pub endpoint_probe_timeout: Option<Duration>,
67}
68
69/// Configuration for serve mode that launches backends on Hello miss.
70#[derive(Clone, Debug)]
71pub struct BrokerLaunchServeConfig {
72    /// Local socket path or Windows pipe name to bind.
73    pub socket_path: String,
74    /// Directory containing `<service>.servicedef` protobuf files.
75    pub service_definition_dir: PathBuf,
76    /// Optional number of control-socket connections to accept before returning.
77    pub max_connections: Option<NonZeroUsize>,
78}
79
80impl BrokerServeConfig {
81    /// Build a serve config using the platform service-definition directory.
82    pub fn new(
83        socket_path: impl Into<String>,
84        service_name: impl Into<String>,
85        service_version: impl Into<String>,
86        backend_endpoint: impl Into<String>,
87        max_connections: usize,
88    ) -> Result<Self, BrokerServeError> {
89        Ok(Self {
90            socket_path: socket_path.into(),
91            service_name: service_name.into(),
92            service_version: service_version.into(),
93            backend_endpoint: backend_endpoint.into(),
94            service_definition_dir: service_definition_dir(),
95            max_connections: Some(
96                NonZeroUsize::new(max_connections)
97                    .ok_or(BrokerServeError::InvalidMaxConnections)?,
98            ),
99            handoff_endpoint: None,
100            endpoint_probe_timeout: None,
101        })
102    }
103
104    /// Build an unbounded serve config using the platform service-definition
105    /// directory.
106    pub fn unbounded(
107        socket_path: impl Into<String>,
108        service_name: impl Into<String>,
109        service_version: impl Into<String>,
110        backend_endpoint: impl Into<String>,
111    ) -> Self {
112        Self {
113            socket_path: socket_path.into(),
114            service_name: service_name.into(),
115            service_version: service_version.into(),
116            backend_endpoint: backend_endpoint.into(),
117            service_definition_dir: service_definition_dir(),
118            max_connections: None,
119            handoff_endpoint: None,
120            endpoint_probe_timeout: None,
121        }
122    }
123
124    /// Override the service-definition directory.
125    pub fn with_service_definition_dir(mut self, root: impl Into<PathBuf>) -> Self {
126        self.service_definition_dir = root.into();
127        self
128    }
129
130    /// Opt in to the Phase 6 handle-passing handoff by configuring the
131    /// backend handoff endpoint the broker dials after negotiation (#387).
132    pub fn with_handoff_endpoint(mut self, endpoint: impl Into<String>) -> Self {
133        self.handoff_endpoint = Some(endpoint.into());
134        self
135    }
136
137    /// Set the startup identity-probe deadline for this serve run.
138    ///
139    /// See [`Self::endpoint_probe_timeout`].
140    pub fn with_endpoint_probe_timeout(mut self, timeout: Duration) -> Self {
141        self.endpoint_probe_timeout = Some(timeout);
142        self
143    }
144
145    /// Return the configured accept-loop connection limit.
146    pub fn connection_limit(&self) -> ControlSocketConnectionLimit {
147        self.max_connections.map_or(
148            ControlSocketConnectionLimit::Unbounded,
149            ControlSocketConnectionLimit::Bounded,
150        )
151    }
152}
153
154impl BrokerLaunchServeConfig {
155    /// Build a launch-backed serve config using the platform
156    /// service-definition directory.
157    pub fn new(
158        socket_path: impl Into<String>,
159        max_connections: usize,
160    ) -> Result<Self, BrokerServeError> {
161        Ok(Self {
162            socket_path: socket_path.into(),
163            service_definition_dir: service_definition_dir(),
164            max_connections: Some(
165                NonZeroUsize::new(max_connections)
166                    .ok_or(BrokerServeError::InvalidMaxConnections)?,
167            ),
168        })
169    }
170
171    /// Build an unbounded launch-backed serve config using the platform
172    /// service-definition directory.
173    pub fn unbounded(socket_path: impl Into<String>) -> Self {
174        Self {
175            socket_path: socket_path.into(),
176            service_definition_dir: service_definition_dir(),
177            max_connections: None,
178        }
179    }
180
181    /// Override the service-definition directory.
182    pub fn with_service_definition_dir(mut self, root: impl Into<PathBuf>) -> Self {
183        self.service_definition_dir = root.into();
184        self
185    }
186
187    /// Return the configured accept-loop connection limit.
188    pub fn connection_limit(&self) -> ControlSocketConnectionLimit {
189        self.max_connections.map_or(
190            ControlSocketConnectionLimit::Unbounded,
191            ControlSocketConnectionLimit::Bounded,
192        )
193    }
194}
195
196/// Serve a bounded number of broker Hello connections.
197pub fn serve_registered_backend(config: BrokerServeConfig) -> Result<(), BrokerServeError> {
198    let RegisteredServeBackend {
199        loader,
200        registry,
201        instance,
202        ..
203    } = build_registered_backend(&config)?;
204    let registry = Mutex::new(registry);
205    let router = HelloRouter::with_lifecycle_monitor(&loader, &registry);
206    let peer_policy =
207        PeerCredentialPolicy::current_user().ok_or(BrokerServeError::PeerPolicyUnavailable)?;
208    let started_at = Instant::now();
209    let fd_guard = FdPressureGuard::default();
210    let snapshot_provider = || {
211        let registry = registry
212            .lock()
213            .unwrap_or_else(|poisoned| poisoned.into_inner());
214        let demoted = fd_guard.is_demoted();
215        AdminSnapshot::from_registry(
216            instance.id(),
217            started_at.elapsed(),
218            !demoted,
219            0,
220            &registry,
221            &[],
222        )
223        .with_fd_pressure_demoted(demoted)
224    };
225    serve_control_socket_connections_with_limit_policy_post_hello_opaque(
226        &config.socket_path,
227        &router,
228        snapshot_provider,
229        config.connection_limit(),
230        &peer_policy,
231        |mut stream, reply| {
232            // Off by default: no handoff endpoint means no handoff attempt.
233            let Some(handoff_endpoint) = config.handoff_endpoint.as_deref() else {
234                return;
235            };
236            let ctx = ServeHandoffContext {
237                handoff_endpoint,
238                service_name: &config.service_name,
239                service_version: &config.service_version,
240                instance: &instance,
241                registry: &registry,
242            };
243            let _must_relinquish = try_complete_negotiated_handoff_opaque(&ctx, &mut stream, reply);
244        },
245        &fd_guard,
246    )?;
247    Ok(())
248}
249
250/// Serve a bounded number of broker Hello connections, launching backends on
251/// verified registry misses.
252pub fn serve_launching_backends(config: BrokerLaunchServeConfig) -> Result<(), BrokerServeError> {
253    let launcher = CommandBackendLauncher::for_current_user()?;
254    serve_launching_backends_with_launcher(config, &launcher)
255}
256
257/// Testable launch-backed serve mode with an injected launcher.
258pub fn serve_launching_backends_with_launcher(
259    config: BrokerLaunchServeConfig,
260    launcher: &dyn BackendLauncher,
261) -> Result<(), BrokerServeError> {
262    let loader = CombinedServiceDefinitionLoader::new(&config.service_definition_dir);
263    let registry = Mutex::new(BackendRegistry::new());
264    let spawn_coordinator = Mutex::new(SpawnCoordinator::new());
265    let router = HelloRouter::with_lifecycle_monitor(&loader, &registry)
266        .with_spawn_coordinator(&spawn_coordinator)
267        .with_backend_launcher(launcher);
268    let peer_policy =
269        PeerCredentialPolicy::current_user().ok_or(BrokerServeError::PeerPolicyUnavailable)?;
270    let started_at = Instant::now();
271    let fd_guard = FdPressureGuard::default();
272    let snapshot_provider = || {
273        let registry = registry
274            .lock()
275            .unwrap_or_else(|poisoned| poisoned.into_inner());
276        let demoted = fd_guard.is_demoted();
277        AdminSnapshot::from_registry("launch", started_at.elapsed(), !demoted, 0, &registry, &[])
278            .with_fd_pressure_demoted(demoted)
279    };
280    serve_launch_control_socket_connections_concurrently(
281        &config.socket_path,
282        &router,
283        snapshot_provider,
284        config.connection_limit(),
285        &peer_policy,
286        &fd_guard,
287    )?;
288    Ok(())
289}
290
291/// Build a Hello handler from one service definition and backend endpoint.
292pub fn build_hello_handler(config: &BrokerServeConfig) -> Result<HelloHandler, BrokerServeError> {
293    let registered = build_registered_backend(config)?;
294    let backend = registered
295        .registry
296        .registered_backend_for_any_build(
297            &registered.instance,
298            &registered.service_definition,
299            &config.service_version,
300        )
301        .ok_or(BrokerServeError::RegisteredBackendMissing)?;
302
303    Ok(HelloHandler::new().with_backend(backend)?)
304}
305
306struct RegisteredServeBackend {
307    loader: CombinedServiceDefinitionLoader,
308    registry: BackendRegistry,
309    instance: BrokerInstanceKey,
310    service_definition: ServiceDefinition,
311}
312
313fn build_registered_backend(
314    config: &BrokerServeConfig,
315) -> Result<RegisteredServeBackend, BrokerServeError> {
316    if config.backend_endpoint.is_empty() {
317        return Err(BrokerServeError::EmptyBackendEndpoint);
318    }
319
320    let loader = CombinedServiceDefinitionLoader::new(&config.service_definition_dir);
321    let service_definition = loader.lookup_or_reload(&config.service_name)?;
322    check_version_allowed(&config.service_version, &service_definition)
323        .map_err(BrokerServeError::VersionPolicy)?;
324
325    let instance = BrokerInstanceKey::from_service_definition(&service_definition)?;
326    let endpoint = Endpoint {
327        namespace_id: instance.id(),
328        path: config.backend_endpoint.clone(),
329    };
330    let daemon = DaemonProcess::current_process(endpoint.clone(), Some(30))?;
331    let handle = BackendHandle::probe_with_service_and_timeout(
332        config.service_name.clone(),
333        config.service_version.clone(),
334        &endpoint,
335        &daemon,
336        config
337            .endpoint_probe_timeout
338            .unwrap_or(crate::broker::backend_lifecycle::probe::DEFAULT_ENDPOINT_PROBE_TIMEOUT),
339    )?;
340
341    let mut registry = BackendRegistry::new();
342    registry.insert(instance.clone(), handle);
343
344    Ok(RegisteredServeBackend {
345        loader,
346        registry,
347        instance,
348        service_definition,
349    })
350}
351
352/// Errors raised while wiring or serving the bounded broker.
353#[derive(Debug, thiserror::Error)]
354pub enum BrokerServeError {
355    /// The connection bound must be non-zero.
356    #[error("max_connections must be greater than zero")]
357    InvalidMaxConnections,
358    /// The configured backend endpoint is empty.
359    #[error("backend endpoint must not be empty")]
360    EmptyBackendEndpoint,
361    /// Service-definition load or validation failed.
362    #[error(transparent)]
363    ServiceDefinition(#[from] ServiceDefinitionError),
364    /// Service isolation could not be mapped to a broker instance.
365    #[error(transparent)]
366    BrokerInstance(#[from] BrokerInstanceError),
367    /// Current process identity could not be recorded for the configured backend.
368    #[error(transparent)]
369    Identity(#[from] IdentityError),
370    /// Current user SID hash could not be computed for backend endpoint allocation.
371    #[error(transparent)]
372    Sid(#[from] SidError),
373    /// Configured backend version is blocked by the service definition.
374    #[error("configured service version is blocked by service-definition policy: {0:?}")]
375    VersionPolicy(VersionPolicyBlock),
376    /// Backend identity verification failed.
377    #[error(transparent)]
378    BackendHandle(#[from] BackendHandleError),
379    /// Registry lookup failed after inserting the configured backend.
380    #[error("registered backend was missing after registry insert")]
381    RegisteredBackendMissing,
382    /// Hello handler construction failed.
383    #[error(transparent)]
384    HelloHandler(#[from] HelloHandlerError),
385    /// The platform current-user peer policy could not be constructed.
386    #[error("current-user peer credential policy is unavailable")]
387    PeerPolicyUnavailable,
388    /// Local-socket serving failed.
389    #[error(transparent)]
390    Connection(#[from] BrokerConnectionError),
391    /// Shared control-socket serving failed.
392    #[error(transparent)]
393    ControlSocket(#[from] ControlSocketError),
394}
running_process/broker/server/serve.rs

running_process/broker/server/
serve.rs
