running-process 4.10.13

//! One-call broker adoption: negotiate → dial → ready-to-talk client (#433 R1).
//!
//! [`connect_to_backend`] returns a raw
//! [`BackendConnection`] — a bare
//! socket the consumer must still wrap in a [`FrameClient`] before it can send
//! a single request. Every consumer (zccache, soldr, clud, fbuild) repeats the
//! same three lines: check the disable env, call `connect_to_backend`, wrap the
//! stream. [`BrokerSession::adopt`] is that recipe, owned once here so the
//! contract is a single call:
//!
//! ```no_run
//! use running_process::broker::adopt::BrokerSession;
//! use running_process::broker::client::ConnectBackendRequest;
//!
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let request = ConnectBackendRequest::new("broker.sock", "zccache", "1.11.20", "1.11.20");
//! let mut session = BrokerSession::adopt(request)?;
//! let reply = session.request(0x7A63, b"ping".to_vec())?;
//! assert_eq!(reply.payload, b"pong");
//! # Ok(()) }
//! ```
//!
//! The blocking [`BrokerSession`] keeps the frozen v1 path. The async
//! `AsyncBrokerSession` (feature `client-async`, #433 R3) keeps the same public
//! type and one-call recipe while default negotiation uses the validated v2
//! Hello exchange (#532); direct-cache, test-seam, and opt-in handoff policies
//! retain their v1 behavior. All blocking socket work stays on
//! `spawn_blocking`, so no second `AsyncRead`/`AsyncWrite` wire exists.

use crate::broker::backend_sdk::{FrameClient, FrameClientError};
use crate::broker::client::{
    broker_disabled_by_env, connect_to_backend, BackendConnection, BackendConnectionRoute,
    BrokerClientError, BrokerDisableEnvError, ConnectBackendRequest,
};
use crate::broker::protocol::{Frame, Negotiated};

/// A negotiated, dialed, and framed broker backend connection.
///
/// Produced by [`BrokerSession::adopt`]. Wraps the
/// [`BackendConnection`] stream in a
/// [`FrameClient`] so the caller can issue correlated request/response frames
/// immediately, while still exposing how the connection was reached
/// ([`route`](Self::route)), the cacheable [`endpoint`](Self::endpoint), and the
/// broker's [`negotiated`](Self::negotiated) metadata.
pub struct BrokerSession {
    client: FrameClient,
    route: BackendConnectionRoute,
    endpoint: String,
    negotiated: Option<Negotiated>,
}

impl BrokerSession {
    /// Negotiate through the broker and return a ready-to-talk session.
    ///
    /// Honours the canonical escape hatch first: if
    /// `RUNNING_PROCESS_DISABLE=1` is set, this returns
    /// [`AdoptError::BrokerDisabled`] so the consumer falls back to its direct
    /// path instead of silently dialing the broker. An invalid disable value
    /// surfaces as [`AdoptError::DisableEnv`].
    pub fn adopt(request: ConnectBackendRequest<'_>) -> Result<Self, AdoptError> {
        if broker_disabled_by_env()? {
            return Err(AdoptError::BrokerDisabled);
        }
        Ok(Self::from_connection(connect_to_backend(request)?))
    }

    fn from_connection(connection: BackendConnection) -> Self {
        Self {
            client: FrameClient::from_stream(connection.stream),
            route: connection.route,
            endpoint: connection.endpoint,
            negotiated: connection.negotiated,
        }
    }

    /// How the backend connection was reached.
    pub fn route(&self) -> BackendConnectionRoute {
        self.route
    }

    /// Negotiated backend endpoint, suitable as a Hello-skip cache key.
    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }

    /// Broker negotiation metadata, present when the broker path was used.
    pub fn negotiated(&self) -> Option<&Negotiated> {
        self.negotiated.as_ref()
    }

    /// Send one correlated request and await its response frame.
    pub fn request(
        &mut self,
        payload_protocol: u32,
        payload: Vec<u8>,
    ) -> Result<Frame, FrameClientError> {
        self.client.request(payload_protocol, payload)
    }

    /// Borrow the underlying frame client for advanced use.
    pub fn client_mut(&mut self) -> &mut FrameClient {
        &mut self.client
    }

    /// Consume the session and return the owned frame client.
    pub fn into_client(self) -> FrameClient {
        self.client
    }

    /// Consume the session and hand back the live negotiated socket as an
    /// owned OS handle (#720).
    ///
    /// After adoption has driven the broker handshake to completion, a
    /// consumer that wants to stop speaking the FrameV1 request/response wire
    /// and run its own protocol over the same connection calls this to take
    /// ownership of the raw socket. On Unix the result wraps an
    /// `OwnedFd`; the Windows `OwnedHandle` path is deferred, so this returns
    /// `IntoBackendIoError::WindowsUnsupported` there for now.
    ///
    /// Fails with [`IntoBackendIoError::BufferedResidual`] if the frame
    /// reader has buffered response bytes the bare socket would not carry —
    /// which never happens on a freshly adopted session that has issued no
    /// [`request`](Self::request).
    pub fn into_backend_io(self) -> Result<OwnedBackendIo, IntoBackendIoError> {
        let buffered = self.client.buffered_len();
        if buffered != 0 {
            return Err(IntoBackendIoError::BufferedResidual { buffered });
        }
        OwnedBackendIo::from_local_socket_stream(self.client.into_stream())
    }
}

/// A live negotiated backend socket handed back as an owned OS handle (#720).
///
/// Produced by [`BrokerSession::into_backend_io`] /
/// `AsyncBrokerSession::into_backend_io`. On Unix it owns an `OwnedFd` the
/// consumer can wrap in its own transport (e.g.
/// `std::os::unix::net::UnixStream::from`); the Windows `OwnedHandle` path is
/// deferred (#720), so the type is never constructed on Windows.
#[derive(Debug)]
pub struct OwnedBackendIo {
    // The Windows handle path is deferred (#720). The type still exists so the
    // `into_backend_io` signature is platform-stable, but it carries no handle
    // on Windows and is only ever returned as `Err(WindowsUnsupported)`.
    #[cfg(unix)]
    fd: std::os::fd::OwnedFd,
}

impl OwnedBackendIo {
    #[cfg(unix)]
    pub(crate) fn from_local_socket_stream(
        stream: crate::platform::ipc::Stream,
    ) -> Result<Self, IntoBackendIoError> {
        Ok(Self {
            fd: stream.into_owned_fd(),
        })
    }

    #[cfg(windows)]
    pub(crate) fn from_local_socket_stream(
        _stream: crate::platform::ipc::Stream,
    ) -> Result<Self, IntoBackendIoError> {
        Err(IntoBackendIoError::WindowsUnsupported)
    }

    /// Consume and return the raw owned file descriptor.
    #[cfg(unix)]
    pub fn into_owned_fd(self) -> std::os::fd::OwnedFd {
        self.fd
    }
}

#[cfg(unix)]
impl std::os::fd::AsFd for OwnedBackendIo {
    fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
        self.fd.as_fd()
    }
}

/// Errors from [`BrokerSession::into_backend_io`] /
/// `AsyncBrokerSession::into_backend_io`.
#[derive(Debug, thiserror::Error)]
pub enum IntoBackendIoError {
    /// The frame reader still holds buffered response bytes that the bare
    /// socket would not carry, so the raw handle cannot be taken without
    /// losing them.
    #[error(
        "frame client has {buffered} buffered response byte(s); cannot hand off the raw socket without losing them"
    )]
    BufferedResidual {
        /// Number of bytes buffered by the frame reader.
        buffered: usize,
    },
    /// The async frame client was poisoned by a prior request panic, so its
    /// inner blocking client is gone.
    #[cfg(feature = "client-async")]
    #[error("async frame client was poisoned by a prior request panic")]
    Poisoned,
    /// `into_backend_io()` is not yet supported on Windows; the `OwnedHandle`
    /// path is deferred (#720).
    #[cfg(windows)]
    #[error("into_backend_io() is not yet supported on Windows; the OwnedHandle path is deferred (#720)")]
    WindowsUnsupported,
}

/// Errors from [`BrokerSession::adopt`] / `AsyncBrokerSession::adopt`.
#[derive(Debug, thiserror::Error)]
pub enum AdoptError {
    /// `RUNNING_PROCESS_DISABLE=1` is set — the caller should use its direct
    /// (non-broker) path. Not a failure of the broker itself.
    #[error("broker disabled via RUNNING_PROCESS_DISABLE=1; use the direct path")]
    BrokerDisabled,
    /// The disable env var held an invalid value.
    #[error(transparent)]
    DisableEnv(#[from] BrokerDisableEnvError),
    /// Broker negotiation or backend dial failed. Use
    /// [`BrokerClientError::refusal_kind`] to branch on broker refusals.
    #[error(transparent)]
    Connect(#[from] BrokerClientError),
    /// The async adoption worker thread failed to join (panicked or was
    /// cancelled). Only reachable on the `client-async` path.
    #[cfg(feature = "client-async")]
    #[error("async adopt worker failed to join: {0}")]
    AsyncJoin(String),
}

/// Owned inputs for [`AsyncBrokerSession::adopt`] (#433 R3).
///
/// The blocking [`ConnectBackendRequest`] borrows `&str`, which cannot cross a
/// `spawn_blocking` boundary. This owned mirror carries the same fields by
/// value; [`AsyncBrokerSession::adopt`] reconstructs a borrowed
/// [`ConnectBackendRequest`] from it inside the worker thread.
#[cfg(feature = "client-async")]
#[derive(Clone, Debug)]
pub struct OwnedConnectRequest {
    /// Broker pipe/socket endpoint.
    pub broker_endpoint: String,
    /// Logical service name, such as `zccache`.
    pub service_name: String,
    /// Backend version the caller wants.
    pub wanted_version: String,
    /// Version of the caller's own service binary.
    pub self_version: String,
    /// Previously negotiated backend endpoint, if the caller has one.
    pub cached_backend_endpoint: Option<String>,
    /// Informational client version.
    pub client_version: String,
    /// Client library name for diagnostics.
    pub client_lib_name: String,
    /// Client library version for diagnostics.
    pub client_lib_version: String,
    /// Proposed keepalive interval.
    pub client_keepalive_secs: u64,
    /// Opt in to adopting a handed-off backend connection.
    pub adopt_handed_off_connection: bool,
    /// Deadline for the handoff-ready relay when adoption is enabled.
    pub handoff_ready_timeout: std::time::Duration,
}

#[cfg(feature = "client-async")]
impl OwnedConnectRequest {
    /// Build an owned request with running-process defaults.
    pub fn new(
        broker_endpoint: impl Into<String>,
        service_name: impl Into<String>,
        wanted_version: impl Into<String>,
        self_version: impl Into<String>,
    ) -> Self {
        Self {
            broker_endpoint: broker_endpoint.into(),
            service_name: service_name.into(),
            wanted_version: wanted_version.into(),
            self_version: self_version.into(),
            cached_backend_endpoint: None,
            client_version: String::new(),
            client_lib_name: "running-process".to_string(),
            client_lib_version: env!("CARGO_PKG_VERSION").to_string(),
            client_keepalive_secs: 0,
            adopt_handed_off_connection: false,
            handoff_ready_timeout: crate::broker::client::DEFAULT_HANDOFF_READY_TIMEOUT,
        }
    }

    fn as_request(&self) -> ConnectBackendRequest<'_> {
        ConnectBackendRequest {
            broker_endpoint: &self.broker_endpoint,
            service_name: &self.service_name,
            wanted_version: &self.wanted_version,
            self_version: &self.self_version,
            cached_backend_endpoint: self.cached_backend_endpoint.as_deref(),
            client_version: &self.client_version,
            client_lib_name: &self.client_lib_name,
            client_lib_version: &self.client_lib_version,
            client_keepalive_secs: self.client_keepalive_secs,
            adopt_handed_off_connection: self.adopt_handed_off_connection,
            handoff_ready_timeout: self.handoff_ready_timeout,
        }
    }
}

/// Async counterpart of [`BrokerSession`] for tokio daemons (#433 R3).
///
/// Runs negotiation and backend dial on `tokio::task::spawn_blocking`, then
/// wraps the resulting [`FrameClient`] in an [`AsyncFrameClient`] so every
/// later request is `.await`-able without a manual blocking worker at the call
/// site. See [`Self::adopt`] for the v2/default and v1-policy split.
///
/// [`AsyncFrameClient`]: crate::broker::backend_sdk::AsyncFrameClient
#[cfg(feature = "client-async")]
pub struct AsyncBrokerSession {
    client: crate::broker::backend_sdk::AsyncFrameClient,
    route: BackendConnectionRoute,
    endpoint: String,
    negotiated: Option<Negotiated>,
}

#[cfg(feature = "client-async")]
impl AsyncBrokerSession {
    /// Negotiate through the broker on a blocking worker and return a
    /// ready-to-talk async session.
    pub async fn adopt(request: OwnedConnectRequest) -> Result<Self, AdoptError> {
        let joined = tokio::task::spawn_blocking(move || adopt_async_blocking(request))
            .await
            .map_err(|err| AdoptError::AsyncJoin(err.to_string()))?;
        let (route, endpoint, negotiated, client) = joined?;
        Ok(Self {
            client: crate::broker::backend_sdk::AsyncFrameClient::from_blocking(client),
            route,
            endpoint,
            negotiated,
        })
    }

    /// How the backend connection was reached.
    pub fn route(&self) -> BackendConnectionRoute {
        self.route
    }

    /// Negotiated backend endpoint, suitable as a Hello-skip cache key.
    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }

    /// Broker negotiation metadata, present when the broker path was used.
    pub fn negotiated(&self) -> Option<&Negotiated> {
        self.negotiated.as_ref()
    }

    /// Send one correlated request and await its response frame.
    pub async fn request(
        &mut self,
        payload_protocol: u32,
        payload: Vec<u8>,
    ) -> Result<Frame, FrameClientError> {
        self.client.request(payload_protocol, payload).await
    }

    /// Consume the session and return the owned async frame client.
    pub fn into_client(self) -> crate::broker::backend_sdk::AsyncFrameClient {
        self.client
    }

    /// Consume the session and hand back the live negotiated socket as an
    /// owned OS handle (#720).
    ///
    /// Async twin of [`BrokerSession::into_backend_io`]. No `.await` is
    /// needed: the inner blocking client already owns the connected socket, so
    /// taking the raw handle out is a synchronous unwrap. Fails with
    /// [`IntoBackendIoError::Poisoned`] if a prior [`request`](Self::request)
    /// panicked inside `spawn_blocking` and left the client slot empty.
    pub fn into_backend_io(self) -> Result<OwnedBackendIo, IntoBackendIoError> {
        let client = self
            .client
            .into_blocking()
            .ok_or(IntoBackendIoError::Poisoned)?;
        let buffered = client.buffered_len();
        if buffered != 0 {
            return Err(IntoBackendIoError::BufferedResidual { buffered });
        }
        OwnedBackendIo::from_local_socket_stream(client.into_stream())
    }
}

#[cfg(feature = "client-async")]
type AdoptedAsync = (
    BackendConnectionRoute,
    String,
    Option<Negotiated>,
    FrameClient,
);

/// Blocking half of async adoption.
///
/// The public async session retains its canonical type identity. Default
/// broker negotiation uses client_v2's validated Hello exchange; the frozen
/// direct-cache, fake-backend, and opt-in handoff paths retain their exact v1
/// behavior because they are transport policies beyond a plain Hello.
#[cfg(feature = "client-async")]
fn adopt_async_blocking(request: OwnedConnectRequest) -> Result<AdoptedAsync, AdoptError> {
    if broker_disabled_by_env()? {
        return Err(AdoptError::BrokerDisabled);
    }

    #[cfg(feature = "test-seams")]
    if std::env::var_os(crate::broker::client::RUNNING_PROCESS_FAKE_BACKEND_ENV)
        .is_some_and(|value| !value.is_empty())
    {
        return BrokerSession::adopt(request.as_request()).map(|session| {
            (
                session.route,
                session.endpoint,
                session.negotiated,
                session.client,
            )
        });
    }

    if request.adopt_handed_off_connection {
        return BrokerSession::adopt(request.as_request()).map(|session| {
            (
                session.route,
                session.endpoint,
                session.negotiated,
                session.client,
            )
        });
    }

    if request.wanted_version == request.self_version {
        if let Some(endpoint) = request.cached_backend_endpoint.as_deref() {
            if let Ok(stream) = crate::broker::client::connect_local_socket(endpoint) {
                return Ok((
                    BackendConnectionRoute::HelloSkip,
                    endpoint.to_owned(),
                    None,
                    FrameClient::from_stream(stream),
                ));
            }
        }
    }

    let mut hello = request.as_request().hello();
    hello.request_id = format!("client_v2-{}-{}", request.service_name, std::process::id());
    let session = crate::broker::client_v2::connect_hello_at_endpoint_with_deadline(
        request.broker_endpoint,
        hello,
        crate::broker::client::broker_client_deadline(),
    )
    .map_err(map_explicit_hello_error)?;
    let negotiated = session.negotiated().clone();
    let endpoint = negotiated.backend_pipe.clone();
    let stream = session
        .connect_backend_ipc()
        .map_err(map_v2_backend_error)?;
    Ok((
        BackendConnectionRoute::BrokerNegotiated,
        endpoint,
        Some(negotiated),
        FrameClient::from_stream(stream),
    ))
}

#[cfg(feature = "client-async")]
fn map_v2_broker_error(error: crate::broker::client_v2::BrokerV2Error) -> AdoptError {
    use crate::broker::client_v2::BrokerV2Error;
    let mapped = match error {
        BrokerV2Error::Dial { source, .. } | BrokerV2Error::Io(source) => {
            BrokerClientError::BrokerConnect(source)
        }
        BrokerV2Error::Framing(source) => BrokerClientError::Framing(source),
        BrokerV2Error::Decode(source) => BrokerClientError::DecodeHelloReply(source),
        BrokerV2Error::MissingResult => BrokerClientError::MissingHelloReplyResult,
        BrokerV2Error::Refused {
            reason,
            retry_after_ms,
            details,
        } => BrokerClientError::Refused {
            code: details.code(),
            reason,
            retry_after_ms,
        },
        other => BrokerClientError::BrokerConnect(std::io::Error::other(other.to_string())),
    };
    AdoptError::Connect(mapped)
}

#[cfg(feature = "client-async")]
fn map_explicit_hello_error(error: crate::broker::client_v2::ExplicitHelloError) -> AdoptError {
    use crate::broker::client_v2::ExplicitHelloError;
    match error {
        ExplicitHelloError::Broker(error) => map_v2_broker_error(error),
        ExplicitHelloError::DecodeFrame(source) => {
            AdoptError::Connect(BrokerClientError::DecodeFrame(source))
        }
        ExplicitHelloError::UnexpectedResponseFrame(reason) => {
            AdoptError::Connect(BrokerClientError::UnexpectedResponseFrame(reason))
        }
    }
}

#[cfg(feature = "client-async")]
fn map_v2_backend_error(error: crate::broker::client_v2::BackendDialError) -> AdoptError {
    use crate::broker::client_v2::BackendDialError;
    let mapped = match error {
        BackendDialError::EmptyBackendPipe => BrokerClientError::EmptyBackendPipe,
        BackendDialError::Connect(source) => BrokerClientError::BackendConnect(source),
        BackendDialError::IntoBackendIo(source) => {
            BrokerClientError::BackendConnect(std::io::Error::other(source.to_string()))
        }
    };
    AdoptError::Connect(mapped)
}