openlatch-client 0.3.3

//! Hot-swappable per-consumer clients — the handle set the self-heal pass writes.
//!
//! The daemon builds every outbound client **once**, at start-up, and hands it to a
//! supervised task that then owns it for the process lifetime. That is the whole reason
//! this module exists: persisting a newly discovered proxy into `config.toml` would leave
//! every live connection pool talking to the dead one until the next restart. D-18 words
//! the requirement as *applied, not just persisted*, and applying it means replacing the
//! client each consumer reads.
//!
//! # Shape
//!
//! One [`ClientHandle`] per consumer, each an `ArcSwap` the consumer re-reads at a natural
//! batch boundary. A swap is a single atomic store; a read is a single atomic load plus a
//! refcount bump, because `reqwest::Client` is itself an `Arc`-backed facade. No task is
//! restarted, so nothing interacts with the supervisor and no in-flight work is dropped.
//!
//! **REJECTED — restarting the supervised tasks to force a rebuild.** The supervisor has no
//! external-restart API, and a restart drops whatever the task was holding (the cloud
//! worker's buffered batch, the poller's in-flight request).
//!
//! **REJECTED — a process-global mutable configuration.** It hides the coupling; the handle
//! set makes the swap surface enumerable, which is what the load-point tests assert against.
//!
//! # Why the client is optional
//!
//! `None` is *"no route is permitted right now"*, and it is the only representation in
//! which `[proxy] allow_direct = false` cannot silently become a direct connection: there
//! is no client to make one with. Every load point has to confront it, and each one has a
//! correct answer (spool, skip the poll, answer 502). See [`crate::error::ERR_DIRECT_FORBIDDEN`].

use std::sync::Arc;

use arc_swap::ArcSwap;

use crate::core::error::OlError;

use super::config::EgressConfig;
use super::factory::{build_client_with, Consumer, Timeouts};

/// One consumer's hot-swappable client.
///
/// Cheap to clone (one refcount bump) and shared by every holder, so the daemon can hand
/// the same handle to a task factory and keep one for the monitor to write.
#[derive(Debug, Clone)]
pub struct ClientHandle(Arc<ArcSwap<Option<reqwest::Client>>>);

impl ClientHandle {
    /// A handle holding `client`, or holding no route when it is `None`.
    pub fn new(client: Option<reqwest::Client>) -> Self {
        Self(Arc::new(ArcSwap::from_pointee(client)))
    }

    /// A handle holding a client.
    pub fn of(client: reqwest::Client) -> Self {
        Self::new(Some(client))
    }

    /// A handle holding no route.
    pub fn empty() -> Self {
        Self::new(None)
    }

    /// The client currently installed, or `None` when no route is permitted.
    ///
    /// **This is the load point.** Call it at a batch boundary — per flush, per poll, per
    /// forwarded request — and use the returned value for that unit of work only. Caching
    /// it in a local that outlives the boundary is exactly the bug the handle exists to
    /// prevent.
    pub fn current(&self) -> Option<reqwest::Client> {
        (**self.0.load()).clone()
    }

    /// This handle, but only while a route is installed.
    ///
    /// The gate the daemon's spawn sites use. A consumer whose client could not be built
    /// at start-up is not started at all — exactly as before these handles existed — while
    /// one that is started keeps re-reading through [`Self::current`].
    pub fn if_routed(&self) -> Option<Self> {
        self.current().map(|_| self.clone())
    }

    /// Install `client` for every subsequent load.
    fn store(&self, client: Option<reqwest::Client>) {
        self.0.store(Arc::new(client));
    }
}

/// Every client the daemon's long-lived consumers send through.
///
/// Built once at start-up and read through [`ClientHandle::current`] by four consumers:
/// the cloud worker (per flush, per drain pass, per health probe), the policy poller (per
/// poll), the alerts long poll (per iteration) and the model boundary (per forwarded
/// request). Editing this set without editing those load points would make the swap apply
/// to nothing.
#[derive(Debug, Clone)]
pub struct EgressClients {
    /// The cloud event worker's client.
    pub cloud: ClientHandle,
    /// The policy-bundle poller's client.
    pub poller: ClientHandle,
    /// The pending-alerts long poll's client.
    pub alerts: ClientHandle,
    /// The model boundary's upstream forwarder.
    pub boundary: ClientHandle,
    /// The cloud worker's per-deployment deadlines, replayed on every rebuild so a
    /// swapped-in client keeps the timeouts the operator configured.
    cloud_timeouts: Timeouts,
    /// The two pollers' shared deadline, replayed for the same reason.
    poll_timeouts: Timeouts,
}

/// Clients built on a candidate route but not yet installed.
///
/// The separation is the contract: a candidate is probed with a *real request* through
/// [`Self::cloud`] and only then handed to [`EgressClients::install`]. That is what makes
/// "`ok` only after a request succeeds through the new route" true by construction rather
/// than by convention.
#[derive(Debug)]
pub struct StagedClients {
    cloud: reqwest::Client,
    poller: reqwest::Client,
    alerts: reqwest::Client,
    boundary: reqwest::Client,
}

impl StagedClients {
    /// The client to probe the candidate with — the same one that will carry the
    /// highest-volume traffic if the candidate wins.
    pub fn cloud(&self) -> &reqwest::Client {
        &self.cloud
    }
}

impl EgressClients {
    /// An empty handle set carrying the deadlines a later build will replay.
    ///
    /// The daemon fills the handles at the sites that already own each consumer's error
    /// handling, so a build failure keeps producing exactly the log line it produces today.
    pub fn new(cloud_timeouts: Timeouts, poll_timeouts: Timeouts) -> Self {
        Self {
            cloud: ClientHandle::empty(),
            poller: ClientHandle::empty(),
            alerts: ClientHandle::empty(),
            boundary: ClientHandle::empty(),
            cloud_timeouts,
            poll_timeouts,
        }
    }

    /// Build one client per consumer on `cfg` without installing any of them.
    ///
    /// All four fail or succeed together: everything [`build_client_with`] rejects is a
    /// property of the configuration, not of the consumer. A partial swap would leave the
    /// daemon straddling two routes, which is worse than not switching at all.
    pub fn stage(&self, cfg: &EgressConfig) -> Result<StagedClients, OlError> {
        Ok(StagedClients {
            cloud: build_client_with(Consumer::CloudWorker, cfg, self.cloud_timeouts)?,
            poller: build_client_with(Consumer::PolicyPoller, cfg, self.poll_timeouts)?,
            alerts: build_client_with(Consumer::Alerts, cfg, self.poll_timeouts)?,
            boundary: build_client_with(Consumer::Boundary, cfg, Timeouts::default())?,
        })
    }

    /// Install a staged route across every consumer.
    ///
    /// The four stores happen back to back with no `await` between them, so no consumer
    /// can observe a mix of two routes for longer than it takes to run four atomic stores.
    pub fn install(&self, staged: StagedClients) {
        self.cloud.store(Some(staged.cloud));
        self.poller.store(Some(staged.poller));
        self.alerts.store(Some(staged.alerts));
        self.boundary.store(Some(staged.boundary));
    }

    /// Build and install `cfg` in one step. Nothing is installed if any client fails.
    pub fn apply(&self, cfg: &EgressConfig) -> Result<(), OlError> {
        self.install(self.stage(cfg)?);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::egress::config::ProxyToml;
    use crate::core::error::ERR_DIRECT_FORBIDDEN;

    struct NoEnv;
    impl crate::core::egress::config::EnvSource for NoEnv {
        fn var(&self, _key: &str) -> Option<String> {
            None
        }
    }

    fn resolve(toml: ProxyToml) -> EgressConfig {
        EgressConfig::resolve(Some(&toml), &NoEnv, 7443, 7444).expect("resolve")
    }

    fn proxied(url: &str) -> EgressConfig {
        resolve(ProxyToml {
            mode: Some("manual".into()),
            url: Some(url.into()),
            source: Some("windows".into()),
            ..Default::default()
        })
    }

    #[test]
    fn an_empty_handle_has_no_client() {
        assert!(ClientHandle::empty().current().is_none());
        assert!(
            ClientHandle::empty().if_routed().is_none(),
            "a consumer must not be started against a handle with no route"
        );
    }

    #[test]
    fn a_swap_is_visible_to_every_holder_of_the_handle() {
        // The property the whole module rests on: the daemon keeps one clone and the
        // consumer keeps another, and a store through either is seen through both.
        let clients = EgressClients::new(Timeouts::default(), Timeouts::default());
        let consumer_side = clients.cloud.clone();
        assert!(consumer_side.current().is_none());

        clients
            .apply(&proxied("http://proxy.test:8080"))
            .expect("apply");
        assert!(
            consumer_side.current().is_some(),
            "a swap must be visible through a handle cloned BEFORE the swap"
        );
    }

    #[test]
    fn every_consumer_handle_is_filled_by_one_apply() {
        let clients = EgressClients::new(Timeouts::default(), Timeouts::default());
        clients.apply(&EgressConfig::direct()).expect("apply");
        for (name, handle) in [
            ("cloud", &clients.cloud),
            ("poller", &clients.poller),
            ("alerts", &clients.alerts),
            ("boundary", &clients.boundary),
        ] {
            assert!(
                handle.current().is_some(),
                "{name} was left without a client"
            );
        }
    }

    #[test]
    fn a_route_that_cannot_be_built_installs_nothing() {
        // `allow_direct = false` with no proxy is the case that must never end at a
        // direct client: `stage` refuses, so there is nothing to install.
        let clients = EgressClients::new(Timeouts::default(), Timeouts::default());
        clients.apply(&EgressConfig::direct()).expect("seed");
        let before = clients.cloud.current().is_some();

        let forbidden = resolve(ProxyToml {
            allow_direct: Some(false),
            ..Default::default()
        });
        let err = clients.apply(&forbidden).expect_err("must refuse");
        assert_eq!(err.code, ERR_DIRECT_FORBIDDEN);
        assert_eq!(
            clients.cloud.current().is_some(),
            before,
            "a refused route must leave the installed one untouched"
        );
    }

    #[test]
    fn staged_clients_are_not_installed_until_install_runs() {
        let clients = EgressClients::new(Timeouts::default(), Timeouts::default());
        let staged = clients
            .stage(&proxied("http://proxy.test:8080"))
            .expect("stage");
        assert!(
            clients.cloud.current().is_none(),
            "staging must not touch the live handles -- the probe runs first"
        );
        clients.install(staged);
        assert!(clients.cloud.current().is_some());
    }
}