running_process/broker/
adopt.rs

1//! One-call broker adoption: negotiate → dial → ready-to-talk client (#433 R1).
2//!
3//! [`connect_to_backend`] returns a raw
4//! [`BackendConnection`] — a bare
5//! socket the consumer must still wrap in a [`FrameClient`] before it can send
6//! a single request. Every consumer (zccache, soldr, clud, fbuild) repeats the
7//! same three lines: check the disable env, call `connect_to_backend`, wrap the
8//! stream. [`BrokerSession::adopt`] is that recipe, owned once here so the
9//! contract is a single call:
10//!
11//! ```no_run
12//! use running_process::broker::adopt::BrokerSession;
13//! use running_process::broker::client::ConnectBackendRequest;
14//!
15//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
16//! let request = ConnectBackendRequest::new("broker.sock", "zccache", "1.11.20", "1.11.20");
17//! let mut session = BrokerSession::adopt(request)?;
18//! let reply = session.request(0x7A63, b"ping".to_vec())?;
19//! assert_eq!(reply.payload, b"pong");
20//! # Ok(()) }
21//! ```
22//!
23//! The blocking [`BrokerSession`] keeps the frozen v1 path. The async
24//! `AsyncBrokerSession` (feature `client-async`, #433 R3) keeps the same public
25//! type and one-call recipe while default negotiation uses the validated v2
26//! Hello exchange (#532); direct-cache, test-seam, and opt-in handoff policies
27//! retain their v1 behavior. All blocking socket work stays on
28//! `spawn_blocking`, so no second `AsyncRead`/`AsyncWrite` wire exists.
29
30use crate::broker::backend_sdk::{FrameClient, FrameClientError};
31use crate::broker::client::{
32    broker_disabled_by_env, connect_to_backend, BackendConnection, BackendConnectionRoute,
33    BrokerClientError, BrokerDisableEnvError, ConnectBackendRequest,
34};
35use crate::broker::protocol::{Frame, Negotiated};
36
37/// A negotiated, dialed, and framed broker backend connection.
38///
39/// Produced by [`BrokerSession::adopt`]. Wraps the
40/// [`BackendConnection`] stream in a
41/// [`FrameClient`] so the caller can issue correlated request/response frames
42/// immediately, while still exposing how the connection was reached
43/// ([`route`](Self::route)), the cacheable [`endpoint`](Self::endpoint), and the
44/// broker's [`negotiated`](Self::negotiated) metadata.
45pub struct BrokerSession {
46    client: FrameClient,
47    route: BackendConnectionRoute,
48    endpoint: String,
49    negotiated: Option<Negotiated>,
50}
51
52impl BrokerSession {
53    /// Negotiate through the broker and return a ready-to-talk session.
54    ///
55    /// Honours the canonical escape hatch first: if
56    /// `RUNNING_PROCESS_DISABLE=1` is set, this returns
57    /// [`AdoptError::BrokerDisabled`] so the consumer falls back to its direct
58    /// path instead of silently dialing the broker. An invalid disable value
59    /// surfaces as [`AdoptError::DisableEnv`].
60    pub fn adopt(request: ConnectBackendRequest<'_>) -> Result<Self, AdoptError> {
61        if broker_disabled_by_env()? {
62            return Err(AdoptError::BrokerDisabled);
63        }
64        Ok(Self::from_connection(connect_to_backend(request)?))
65    }
66
67    fn from_connection(connection: BackendConnection) -> Self {
68        Self {
69            client: FrameClient::from_stream(connection.stream),
70            route: connection.route,
71            endpoint: connection.endpoint,
72            negotiated: connection.negotiated,
73        }
74    }
75
76    /// How the backend connection was reached.
77    pub fn route(&self) -> BackendConnectionRoute {
78        self.route
79    }
80
81    /// Negotiated backend endpoint, suitable as a Hello-skip cache key.
82    pub fn endpoint(&self) -> &str {
83        &self.endpoint
84    }
85
86    /// Broker negotiation metadata, present when the broker path was used.
87    pub fn negotiated(&self) -> Option<&Negotiated> {
88        self.negotiated.as_ref()
89    }
90
91    /// Send one correlated request and await its response frame.
92    pub fn request(
93        &mut self,
94        payload_protocol: u32,
95        payload: Vec<u8>,
96    ) -> Result<Frame, FrameClientError> {
97        self.client.request(payload_protocol, payload)
98    }
99
100    /// Borrow the underlying frame client for advanced use.
101    pub fn client_mut(&mut self) -> &mut FrameClient {
102        &mut self.client
103    }
104
105    /// Consume the session and return the owned frame client.
106    pub fn into_client(self) -> FrameClient {
107        self.client
108    }
109
110    /// Consume the session and hand back the live negotiated socket as an
111    /// owned OS handle (#720).
112    ///
113    /// After adoption has driven the broker handshake to completion, a
114    /// consumer that wants to stop speaking the FrameV1 request/response wire
115    /// and run its own protocol over the same connection calls this to take
116    /// ownership of the raw socket. On Unix the result wraps an
117    /// `OwnedFd`; the Windows `OwnedHandle` path is deferred, so this returns
118    /// `IntoBackendIoError::WindowsUnsupported` there for now.
119    ///
120    /// Fails with [`IntoBackendIoError::BufferedResidual`] if the frame
121    /// reader has buffered response bytes the bare socket would not carry —
122    /// which never happens on a freshly adopted session that has issued no
123    /// [`request`](Self::request).
124    pub fn into_backend_io(self) -> Result<OwnedBackendIo, IntoBackendIoError> {
125        let buffered = self.client.buffered_len();
126        if buffered != 0 {
127            return Err(IntoBackendIoError::BufferedResidual { buffered });
128        }
129        OwnedBackendIo::from_local_socket_stream(self.client.into_stream())
130    }
131}
132
133/// A live negotiated backend socket handed back as an owned OS handle (#720).
134///
135/// Produced by [`BrokerSession::into_backend_io`] /
136/// `AsyncBrokerSession::into_backend_io`. On Unix it owns an `OwnedFd` the
137/// consumer can wrap in its own transport (e.g.
138/// `std::os::unix::net::UnixStream::from`); the Windows `OwnedHandle` path is
139/// deferred (#720), so the type is never constructed on Windows.
140#[derive(Debug)]
141pub struct OwnedBackendIo {
142    // The Windows handle path is deferred (#720). The type still exists so the
143    // `into_backend_io` signature is platform-stable, but it carries no handle
144    // on Windows and is only ever returned as `Err(WindowsUnsupported)`.
145    #[cfg(unix)]
146    fd: std::os::fd::OwnedFd,
147}
148
149impl OwnedBackendIo {
150    #[cfg(unix)]
151    pub(crate) fn from_local_socket_stream(
152        stream: crate::platform::ipc::Stream,
153    ) -> Result<Self, IntoBackendIoError> {
154        Ok(Self {
155            fd: stream.into_owned_fd(),
156        })
157    }
158
159    #[cfg(windows)]
160    pub(crate) fn from_local_socket_stream(
161        _stream: crate::platform::ipc::Stream,
162    ) -> Result<Self, IntoBackendIoError> {
163        Err(IntoBackendIoError::WindowsUnsupported)
164    }
165
166    /// Consume and return the raw owned file descriptor.
167    #[cfg(unix)]
168    pub fn into_owned_fd(self) -> std::os::fd::OwnedFd {
169        self.fd
170    }
171}
172
173#[cfg(unix)]
174impl std::os::fd::AsFd for OwnedBackendIo {
175    fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
176        self.fd.as_fd()
177    }
178}
179
180/// Errors from [`BrokerSession::into_backend_io`] /
181/// `AsyncBrokerSession::into_backend_io`.
182#[derive(Debug, thiserror::Error)]
183pub enum IntoBackendIoError {
184    /// The frame reader still holds buffered response bytes that the bare
185    /// socket would not carry, so the raw handle cannot be taken without
186    /// losing them.
187    #[error(
188        "frame client has {buffered} buffered response byte(s); cannot hand off the raw socket without losing them"
189    )]
190    BufferedResidual {
191        /// Number of bytes buffered by the frame reader.
192        buffered: usize,
193    },
194    /// The async frame client was poisoned by a prior request panic, so its
195    /// inner blocking client is gone.
196    #[cfg(feature = "client-async")]
197    #[error("async frame client was poisoned by a prior request panic")]
198    Poisoned,
199    /// `into_backend_io()` is not yet supported on Windows; the `OwnedHandle`
200    /// path is deferred (#720).
201    #[cfg(windows)]
202    #[error("into_backend_io() is not yet supported on Windows; the OwnedHandle path is deferred (#720)")]
203    WindowsUnsupported,
204}
205
206/// Errors from [`BrokerSession::adopt`] / `AsyncBrokerSession::adopt`.
207#[derive(Debug, thiserror::Error)]
208pub enum AdoptError {
209    /// `RUNNING_PROCESS_DISABLE=1` is set — the caller should use its direct
210    /// (non-broker) path. Not a failure of the broker itself.
211    #[error("broker disabled via RUNNING_PROCESS_DISABLE=1; use the direct path")]
212    BrokerDisabled,
213    /// The disable env var held an invalid value.
214    #[error(transparent)]
215    DisableEnv(#[from] BrokerDisableEnvError),
216    /// Broker negotiation or backend dial failed. Use
217    /// [`BrokerClientError::refusal_kind`] to branch on broker refusals.
218    #[error(transparent)]
219    Connect(#[from] BrokerClientError),
220    /// The async adoption worker thread failed to join (panicked or was
221    /// cancelled). Only reachable on the `client-async` path.
222    #[cfg(feature = "client-async")]
223    #[error("async adopt worker failed to join: {0}")]
224    AsyncJoin(String),
225}
226
227/// Owned inputs for [`AsyncBrokerSession::adopt`] (#433 R3).
228///
229/// The blocking [`ConnectBackendRequest`] borrows `&str`, which cannot cross a
230/// `spawn_blocking` boundary. This owned mirror carries the same fields by
231/// value; [`AsyncBrokerSession::adopt`] reconstructs a borrowed
232/// [`ConnectBackendRequest`] from it inside the worker thread.
233#[cfg(feature = "client-async")]
234#[derive(Clone, Debug)]
235pub struct OwnedConnectRequest {
236    /// Broker pipe/socket endpoint.
237    pub broker_endpoint: String,
238    /// Logical service name, such as `zccache`.
239    pub service_name: String,
240    /// Backend version the caller wants.
241    pub wanted_version: String,
242    /// Version of the caller's own service binary.
243    pub self_version: String,
244    /// Previously negotiated backend endpoint, if the caller has one.
245    pub cached_backend_endpoint: Option<String>,
246    /// Informational client version.
247    pub client_version: String,
248    /// Client library name for diagnostics.
249    pub client_lib_name: String,
250    /// Client library version for diagnostics.
251    pub client_lib_version: String,
252    /// Proposed keepalive interval.
253    pub client_keepalive_secs: u64,
254    /// Opt in to adopting a handed-off backend connection.
255    pub adopt_handed_off_connection: bool,
256    /// Deadline for the handoff-ready relay when adoption is enabled.
257    pub handoff_ready_timeout: std::time::Duration,
258}
259
260#[cfg(feature = "client-async")]
261impl OwnedConnectRequest {
262    /// Build an owned request with running-process defaults.
263    pub fn new(
264        broker_endpoint: impl Into<String>,
265        service_name: impl Into<String>,
266        wanted_version: impl Into<String>,
267        self_version: impl Into<String>,
268    ) -> Self {
269        Self {
270            broker_endpoint: broker_endpoint.into(),
271            service_name: service_name.into(),
272            wanted_version: wanted_version.into(),
273            self_version: self_version.into(),
274            cached_backend_endpoint: None,
275            client_version: String::new(),
276            client_lib_name: "running-process".to_string(),
277            client_lib_version: env!("CARGO_PKG_VERSION").to_string(),
278            client_keepalive_secs: 0,
279            adopt_handed_off_connection: false,
280            handoff_ready_timeout: crate::broker::client::DEFAULT_HANDOFF_READY_TIMEOUT,
281        }
282    }
283
284    fn as_request(&self) -> ConnectBackendRequest<'_> {
285        ConnectBackendRequest {
286            broker_endpoint: &self.broker_endpoint,
287            service_name: &self.service_name,
288            wanted_version: &self.wanted_version,
289            self_version: &self.self_version,
290            cached_backend_endpoint: self.cached_backend_endpoint.as_deref(),
291            client_version: &self.client_version,
292            client_lib_name: &self.client_lib_name,
293            client_lib_version: &self.client_lib_version,
294            client_keepalive_secs: self.client_keepalive_secs,
295            adopt_handed_off_connection: self.adopt_handed_off_connection,
296            handoff_ready_timeout: self.handoff_ready_timeout,
297        }
298    }
299}
300
301/// Async counterpart of [`BrokerSession`] for tokio daemons (#433 R3).
302///
303/// Runs negotiation and backend dial on `tokio::task::spawn_blocking`, then
304/// wraps the resulting [`FrameClient`] in an [`AsyncFrameClient`] so every
305/// later request is `.await`-able without a manual blocking worker at the call
306/// site. See [`Self::adopt`] for the v2/default and v1-policy split.
307///
308/// [`AsyncFrameClient`]: crate::broker::backend_sdk::AsyncFrameClient
309#[cfg(feature = "client-async")]
310pub struct AsyncBrokerSession {
311    client: crate::broker::backend_sdk::AsyncFrameClient,
312    route: BackendConnectionRoute,
313    endpoint: String,
314    negotiated: Option<Negotiated>,
315}
316
317#[cfg(feature = "client-async")]
318impl AsyncBrokerSession {
319    /// Negotiate through the broker on a blocking worker and return a
320    /// ready-to-talk async session.
321    pub async fn adopt(request: OwnedConnectRequest) -> Result<Self, AdoptError> {
322        let joined = tokio::task::spawn_blocking(move || adopt_async_blocking(request))
323            .await
324            .map_err(|err| AdoptError::AsyncJoin(err.to_string()))?;
325        let (route, endpoint, negotiated, client) = joined?;
326        Ok(Self {
327            client: crate::broker::backend_sdk::AsyncFrameClient::from_blocking(client),
328            route,
329            endpoint,
330            negotiated,
331        })
332    }
333
334    /// How the backend connection was reached.
335    pub fn route(&self) -> BackendConnectionRoute {
336        self.route
337    }
338
339    /// Negotiated backend endpoint, suitable as a Hello-skip cache key.
340    pub fn endpoint(&self) -> &str {
341        &self.endpoint
342    }
343
344    /// Broker negotiation metadata, present when the broker path was used.
345    pub fn negotiated(&self) -> Option<&Negotiated> {
346        self.negotiated.as_ref()
347    }
348
349    /// Send one correlated request and await its response frame.
350    pub async fn request(
351        &mut self,
352        payload_protocol: u32,
353        payload: Vec<u8>,
354    ) -> Result<Frame, FrameClientError> {
355        self.client.request(payload_protocol, payload).await
356    }
357
358    /// Consume the session and return the owned async frame client.
359    pub fn into_client(self) -> crate::broker::backend_sdk::AsyncFrameClient {
360        self.client
361    }
362
363    /// Consume the session and hand back the live negotiated socket as an
364    /// owned OS handle (#720).
365    ///
366    /// Async twin of [`BrokerSession::into_backend_io`]. No `.await` is
367    /// needed: the inner blocking client already owns the connected socket, so
368    /// taking the raw handle out is a synchronous unwrap. Fails with
369    /// [`IntoBackendIoError::Poisoned`] if a prior [`request`](Self::request)
370    /// panicked inside `spawn_blocking` and left the client slot empty.
371    pub fn into_backend_io(self) -> Result<OwnedBackendIo, IntoBackendIoError> {
372        let client = self
373            .client
374            .into_blocking()
375            .ok_or(IntoBackendIoError::Poisoned)?;
376        let buffered = client.buffered_len();
377        if buffered != 0 {
378            return Err(IntoBackendIoError::BufferedResidual { buffered });
379        }
380        OwnedBackendIo::from_local_socket_stream(client.into_stream())
381    }
382}
383
384#[cfg(feature = "client-async")]
385type AdoptedAsync = (
386    BackendConnectionRoute,
387    String,
388    Option<Negotiated>,
389    FrameClient,
390);
391
392/// Blocking half of async adoption.
393///
394/// The public async session retains its canonical type identity. Default
395/// broker negotiation uses client_v2's validated Hello exchange; the frozen
396/// direct-cache, fake-backend, and opt-in handoff paths retain their exact v1
397/// behavior because they are transport policies beyond a plain Hello.
398#[cfg(feature = "client-async")]
399fn adopt_async_blocking(request: OwnedConnectRequest) -> Result<AdoptedAsync, AdoptError> {
400    if broker_disabled_by_env()? {
401        return Err(AdoptError::BrokerDisabled);
402    }
403
404    #[cfg(feature = "test-seams")]
405    if std::env::var_os(crate::broker::client::RUNNING_PROCESS_FAKE_BACKEND_ENV)
406        .is_some_and(|value| !value.is_empty())
407    {
408        return BrokerSession::adopt(request.as_request()).map(|session| {
409            (
410                session.route,
411                session.endpoint,
412                session.negotiated,
413                session.client,
414            )
415        });
416    }
417
418    if request.adopt_handed_off_connection {
419        return BrokerSession::adopt(request.as_request()).map(|session| {
420            (
421                session.route,
422                session.endpoint,
423                session.negotiated,
424                session.client,
425            )
426        });
427    }
428
429    if request.wanted_version == request.self_version {
430        if let Some(endpoint) = request.cached_backend_endpoint.as_deref() {
431            if let Ok(stream) = crate::broker::client::connect_local_socket(endpoint) {
432                return Ok((
433                    BackendConnectionRoute::HelloSkip,
434                    endpoint.to_owned(),
435                    None,
436                    FrameClient::from_stream(stream),
437                ));
438            }
439        }
440    }
441
442    let mut hello = request.as_request().hello();
443    hello.request_id = format!("client_v2-{}-{}", request.service_name, std::process::id());
444    let session = crate::broker::client_v2::connect_hello_at_endpoint_with_deadline(
445        request.broker_endpoint,
446        hello,
447        crate::broker::client::broker_client_deadline(),
448    )
449    .map_err(map_explicit_hello_error)?;
450    let negotiated = session.negotiated().clone();
451    let endpoint = negotiated.backend_pipe.clone();
452    let stream = session
453        .connect_backend_ipc()
454        .map_err(map_v2_backend_error)?;
455    Ok((
456        BackendConnectionRoute::BrokerNegotiated,
457        endpoint,
458        Some(negotiated),
459        FrameClient::from_stream(stream),
460    ))
461}
462
463#[cfg(feature = "client-async")]
464fn map_v2_broker_error(error: crate::broker::client_v2::BrokerV2Error) -> AdoptError {
465    use crate::broker::client_v2::BrokerV2Error;
466    let mapped = match error {
467        BrokerV2Error::Dial { source, .. } | BrokerV2Error::Io(source) => {
468            BrokerClientError::BrokerConnect(source)
469        }
470        BrokerV2Error::Framing(source) => BrokerClientError::Framing(source),
471        BrokerV2Error::Decode(source) => BrokerClientError::DecodeHelloReply(source),
472        BrokerV2Error::MissingResult => BrokerClientError::MissingHelloReplyResult,
473        BrokerV2Error::Refused {
474            reason,
475            retry_after_ms,
476            details,
477        } => BrokerClientError::Refused {
478            code: details.code(),
479            reason,
480            retry_after_ms,
481        },
482        other => BrokerClientError::BrokerConnect(std::io::Error::other(other.to_string())),
483    };
484    AdoptError::Connect(mapped)
485}
486
487#[cfg(feature = "client-async")]
488fn map_explicit_hello_error(error: crate::broker::client_v2::ExplicitHelloError) -> AdoptError {
489    use crate::broker::client_v2::ExplicitHelloError;
490    match error {
491        ExplicitHelloError::Broker(error) => map_v2_broker_error(error),
492        ExplicitHelloError::DecodeFrame(source) => {
493            AdoptError::Connect(BrokerClientError::DecodeFrame(source))
494        }
495        ExplicitHelloError::UnexpectedResponseFrame(reason) => {
496            AdoptError::Connect(BrokerClientError::UnexpectedResponseFrame(reason))
497        }
498    }
499}
500
501#[cfg(feature = "client-async")]
502fn map_v2_backend_error(error: crate::broker::client_v2::BackendDialError) -> AdoptError {
503    use crate::broker::client_v2::BackendDialError;
504    let mapped = match error {
505        BackendDialError::EmptyBackendPipe => BrokerClientError::EmptyBackendPipe,
506        BackendDialError::Connect(source) => BrokerClientError::BackendConnect(source),
507        BackendDialError::IntoBackendIo(source) => {
508            BrokerClientError::BackendConnect(std::io::Error::other(source.to_string()))
509        }
510    };
511    AdoptError::Connect(mapped)
512}
running_process/broker/adopt.rs

running_process/broker/
adopt.rs
