orch/cluster/
local.rs

1//! The localhost endpoint the CLI sends replicated writes to, so they go through the daemon's log
2
3use std::path::PathBuf;
4use std::sync::{Arc, LazyLock};
5use std::time::Duration;
6
7use axum::extract::State;
8use axum::http::{header, HeaderMap, StatusCode};
9use axum::response::{IntoResponse, Response};
10use axum::routing::{get, post};
11use axum::{Json, Router};
12use subtle::ConstantTimeEq;
13
14use super::{Command, Reply};
15use crate::store::TaskStore;
16
17static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
18    reqwest::Client::builder()
19        .timeout(Duration::from_secs(30))
20        .build()
21        .unwrap_or_default()
22});
23
24/// Where the CLI reaches the daemon, and the token that proves the caller can read `~/.orch`
25#[derive(Debug, Clone)]
26pub struct Endpoint {
27    pub url: String,
28    pub token: String,
29}
30
31fn port_path() -> anyhow::Result<PathBuf> {
32    if cfg!(test) {
33        Ok(std::env::temp_dir()
34            .join("orch-test-state")
35            .join("cluster.port"))
36    } else {
37        Ok(crate::home::state_dir()?.join("cluster.port"))
38    }
39}
40
41#[derive(Clone)]
42struct App {
43    store: TaskStore,
44    token: Arc<String>,
45    host: Arc<String>,
46}
47
48/// Serve `POST /write` on 127.0.0.1 and record the port, a fresh token and the pid for the CLI
49pub async fn serve(store: TaskStore) -> anyhow::Result<Endpoint> {
50    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
51    let port = listener.local_addr()?.port();
52    let token = format!(
53        "{}{}",
54        uuid::Uuid::new_v4().simple(),
55        uuid::Uuid::new_v4().simple()
56    );
57    write_private(
58        &port_path()?,
59        &format!("{port} {token} {}", std::process::id()),
60    )?;
61    let app = Router::new()
62        .route("/write", post(write))
63        .route("/progress", post(progress))
64        .route("/nodes", get(nodes))
65        .with_state(App {
66            store,
67            token: Arc::new(token.clone()),
68            host: Arc::new(format!("127.0.0.1:{port}")),
69        });
70    tokio::spawn(async move {
71        if let Err(e) = axum::serve(listener, app).await {
72            tracing::warn!(err = %e, "cluster write endpoint stopped");
73        }
74    });
75    tracing::info!(port, "cluster write endpoint started on 127.0.0.1");
76    Ok(Endpoint {
77        url: format!("http://127.0.0.1:{port}/write"),
78        token,
79    })
80}
81
82/// Owner-only file, written to a temp name and renamed so a reader never sees half of it
83fn write_private(path: &std::path::Path, body: &str) -> anyhow::Result<()> {
84    use std::io::Write;
85    use std::os::unix::fs::OpenOptionsExt;
86    if let Some(parent) = path.parent() {
87        std::fs::create_dir_all(parent)?;
88    }
89
90    let tmp = path.with_extension(format!("tmp.{}", std::process::id()));
91    let mut file = std::fs::OpenOptions::new()
92        .write(true)
93        .create(true)
94        .truncate(true)
95        .mode(0o600)
96        .open(&tmp)?;
97    file.write_all(body.as_bytes())?;
98    std::fs::rename(&tmp, path)?;
99    Ok(())
100}
101
102fn authorized(app: &App, headers: &HeaderMap) -> bool {
103    // A rebound DNS name sends its own Host, so only the literal address passes
104    let host_ok = headers
105        .get(header::HOST)
106        .is_some_and(|h| h.as_bytes() == app.host.as_bytes());
107    let given = headers
108        .get(header::AUTHORIZATION)
109        .and_then(|v| v.to_str().ok())
110        .and_then(|v| v.strip_prefix("Bearer "))
111        .unwrap_or("");
112    let token_ok: bool = given.as_bytes().ct_eq(app.token.as_bytes()).into();
113    host_ok && token_ok
114}
115
116async fn write(State(app): State<App>, headers: HeaderMap, Json(cmd): Json<Command>) -> Response {
117    if !authorized(&app, &headers) {
118        return StatusCode::UNAUTHORIZED.into_response();
119    }
120
121    Json(match app.store.write(cmd).await {
122        Ok(reply) => reply,
123        Err(e) => Reply::Err(format!("{e:#}")),
124    })
125    .into_response()
126}
127
128async fn progress(
129    State(app): State<App>,
130    headers: HeaderMap,
131    Json(report): Json<super::progress::Report>,
132) -> Response {
133    if !authorized(&app, &headers) {
134        return StatusCode::UNAUTHORIZED.into_response();
135    }
136
137    let node = crate::cluster::discovery::status()
138        .await
139        .map(|t| t.me.name)
140        .unwrap_or_else(|_| "local".to_string());
141    match super::progress::handle(&app.store, &node, report).await {
142        Ok(record) => Json(record).into_response(),
143        Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
144    }
145}
146
147async fn nodes(State(app): State<App>, headers: HeaderMap) -> Response {
148    if !authorized(&app, &headers) {
149        return StatusCode::UNAUTHORIZED.into_response();
150    }
151
152    let members = app.store.node().map(super::members).unwrap_or_default();
153    Json(members).into_response()
154}
155
156pub async fn fetch_nodes(endpoint: &Endpoint) -> anyhow::Result<Vec<super::Member>> {
157    let url = endpoint.url.replace("/write", "/nodes");
158    let res = CLIENT.get(url).bearer_auth(&endpoint.token).send().await?;
159    Ok(res.error_for_status()?.json().await?)
160}
161
162/// Send an `orch progress` report to the daemon, which commits, pushes and records it
163pub async fn send_progress(
164    endpoint: &Endpoint,
165    report: &super::progress::Report,
166) -> anyhow::Result<super::progress::Record> {
167    let url = endpoint.url.replace("/write", "/progress");
168    let res = CLIENT
169        .post(url)
170        .bearer_auth(&endpoint.token)
171        .json(report)
172        .send()
173        .await?;
174    let status = res.status();
175    if !status.is_success() {
176        anyhow::bail!(
177            "orch service refused the report ({status}): {}",
178            res.text().await?
179        );
180    }
181
182    Ok(res.json().await?)
183}
184
185pub async fn cleanup() {
186    if let Ok(path) = port_path() {
187        let _ = tokio::fs::remove_file(path).await;
188    }
189}
190
191/// The running daemon's endpoint. A file left by a daemon that is gone names nothing
192pub async fn daemon_endpoint() -> Option<Endpoint> {
193    let body = tokio::fs::read_to_string(port_path().ok()?).await.ok()?;
194    let mut parts = body.split_whitespace();
195    let port: u16 = parts.next()?.parse().ok()?;
196    let token = parts.next()?.to_string();
197    let pid: u32 = parts.next()?.parse().ok()?;
198    let alive = tokio::task::spawn_blocking(move || crate::home::pid_is_orch_serve(pid))
199        .await
200        .ok()?;
201    alive.then(|| Endpoint {
202        url: format!("http://127.0.0.1:{port}/write"),
203        token,
204    })
205}
206
207pub(crate) async fn send(endpoint: &Endpoint, cmd: &Command) -> reqwest::Result<Reply> {
208    CLIENT
209        .post(&endpoint.url)
210        .bearer_auth(&endpoint.token)
211        .json(cmd)
212        .send()
213        .await?
214        .error_for_status()?
215        .json()
216        .await
217}
orch/cluster/local.rs

orch/cluster/
local.rs
