maestro/crates/maestro-cli/src/http.rs

//! Thin HTTP client wrapper around `reqwest` for the daemon's `/api/*` surface.

use std::time::Duration;

use reqwest::{Method, Response, StatusCode};
use serde::de::DeserializeOwned;
use serde::Serialize;

use crate::config::ConfigError;

/// Time budget for establishing a TCP connection to the daemon.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
/// Total time budget for a single request (connect + send + receive). Bounds
/// the case where the peer is TCP-alive but never makes forward progress.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);

/// Build a `reqwest::Client` with the CLI's standard timeouts and user-agent.
///
/// Centralizing construction here guarantees every client the CLI uses gets a
/// connect *and* request timeout, so `maestro` never hangs forever on an
/// unhealthy daemon (NAP-262).
pub fn build_client() -> reqwest::Client {
    build_client_with_timeouts(CONNECT_TIMEOUT, REQUEST_TIMEOUT)
}

fn build_client_with_timeouts(connect: Duration, request: Duration) -> reqwest::Client {
    reqwest::Client::builder()
        .user_agent(concat!("maestro-cli/", env!("CARGO_PKG_VERSION")))
        .connect_timeout(connect)
        .timeout(request)
        .build()
        .expect("build reqwest client")
}

#[derive(Debug, thiserror::Error)]
pub enum HttpError {
    #[error("http: {0}")]
    Reqwest(#[from] reqwest::Error),
    #[error("server returned {status}: {body}")]
    ServerError { status: StatusCode, body: String },
    #[error("decode: {0}")]
    Json(#[from] serde_json::Error),
    #[error("config: {0}")]
    Config(#[from] ConfigError),
}

impl HttpError {
    /// True when the error is a transport-level failure to reach the host
    /// (connection refused / connect error) rather than an HTTP-level
    /// rejection.
    pub fn is_unreachable(&self) -> bool {
        matches!(self, HttpError::Reqwest(e) if e.is_connect())
    }
}

/// HTTP client bound to one daemon's `/api/*` base URL. The daemon's surface
/// needs no auth — reaching its bind address is the authorization.
#[derive(Debug)]
pub struct DaemonClient {
    base: String,
    inner: reqwest::Client,
}

impl DaemonClient {
    pub fn new(base: String) -> Self {
        Self {
            base: base.trim_end_matches('/').to_string(),
            inner: build_client(),
        }
    }

    pub fn base(&self) -> &str {
        &self.base
    }

    async fn send(
        &self,
        method: Method,
        path: &str,
        body: Option<serde_json::Value>,
    ) -> Result<Response, HttpError> {
        let url = format!("{}{}", self.base, path);
        let mut req = self.inner.request(method, &url);
        if let Some(b) = body {
            req = req.json(&b);
        }
        Ok(req.send().await?)
    }

    pub async fn get_json<T: DeserializeOwned>(&self, path: &str) -> Result<T, HttpError> {
        let resp = self.send(Method::GET, path, None).await?;
        ok_or_body(resp)
            .await?
            .json::<T>()
            .await
            .map_err(Into::into)
    }

    pub async fn post_json<B: Serialize, T: DeserializeOwned>(
        &self,
        path: &str,
        body: &B,
    ) -> Result<T, HttpError> {
        let v = serde_json::to_value(body)?;
        let resp = self.send(Method::POST, path, Some(v)).await?;
        ok_or_body(resp)
            .await?
            .json::<T>()
            .await
            .map_err(Into::into)
    }

    pub async fn delete(&self, path: &str) -> Result<(), HttpError> {
        let resp = self.send(Method::DELETE, path, None).await?;
        ok_or_body(resp).await?;
        Ok(())
    }
}

async fn ok_or_body(resp: Response) -> Result<Response, HttpError> {
    let status = resp.status();
    if status.is_success() {
        return Ok(resp);
    }
    let body = resp.text().await.unwrap_or_default();
    Err(HttpError::ServerError { status, body })
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A peer that accepts the TCP connection but never sends a response must
    /// not hang the client forever: the request timeout has to fire (NAP-262).
    #[tokio::test]
    async fn request_timeout_fires_when_peer_never_responds() {
        use std::io::Read;
        use std::net::TcpListener;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        // Accept one connection and sit on it without ever writing a response.
        let handle = std::thread::spawn(move || {
            if let Ok((mut sock, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = sock.read(&mut buf);
                std::thread::sleep(Duration::from_secs(2));
            }
        });

        let client = build_client_with_timeouts(Duration::from_secs(5), Duration::from_millis(300));
        let err = client
            .get(format!("http://{addr}/"))
            .send()
            .await
            .expect_err("request should time out, not hang");
        assert!(err.is_timeout(), "expected timeout error, got: {err}");

        let _ = handle.join();
    }
}