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

orch/cluster/
local.rs
