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::{HeaderMap, StatusCode, header};
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<str>,
45    host: Arc<str>,
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::from(token.as_str()),
68            host: Arc::from(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    // The store writes local state directly, so a command that carries any came from outside it
122    if let Err(e) = cmd.check_replicable() {
123        return Json(Reply::Err(format!("{e:#}"))).into_response();
124    }
125
126    Json(match app.store.write(cmd).await {
127        Ok(reply) => reply,
128        Err(e) => Reply::Err(format!("{e:#}")),
129    })
130    .into_response()
131}
132
133async fn progress(
134    State(app): State<App>,
135    headers: HeaderMap,
136    Json(report): Json<super::progress::Report>,
137) -> Response {
138    if !authorized(&app, &headers) {
139        return StatusCode::UNAUTHORIZED.into_response();
140    }
141
142    let node = crate::cluster::discovery::status()
143        .await
144        .map_or_else(|_| "local".to_string(), |t| t.me.name);
145    match super::progress::handle(&app.store, &node, report).await {
146        Ok(record) => Json(record).into_response(),
147        Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
148    }
149}
150
151async fn nodes(State(app): State<App>, headers: HeaderMap) -> Response {
152    if !authorized(&app, &headers) {
153        return StatusCode::UNAUTHORIZED.into_response();
154    }
155
156    let members = app.store.node().map(super::members).unwrap_or_default();
157    Json(members).into_response()
158}
159
160pub async fn fetch_nodes(endpoint: &Endpoint) -> anyhow::Result<Vec<super::Member>> {
161    let url = endpoint.url.replace("/write", "/nodes");
162    let res = CLIENT.get(url).bearer_auth(&endpoint.token).send().await?;
163    Ok(res.error_for_status()?.json().await?)
164}
165
166/// Send an `orch progress` report to the daemon, which commits, pushes and records it
167pub async fn send_progress(
168    endpoint: &Endpoint,
169    report: &super::progress::Report,
170) -> anyhow::Result<super::progress::Record> {
171    let url = endpoint.url.replace("/write", "/progress");
172    let res = CLIENT
173        .post(url)
174        .bearer_auth(&endpoint.token)
175        .json(report)
176        .send()
177        .await?;
178    let status = res.status();
179    if !status.is_success() {
180        anyhow::bail!(
181            "orch service refused the report ({status}): {}",
182            res.text().await?
183        );
184    }
185
186    Ok(res.json().await?)
187}
188
189pub async fn cleanup() {
190    if let Ok(path) = port_path() {
191        let _ = tokio::fs::remove_file(path).await;
192    }
193}
194
195/// The running daemon's endpoint. A file left by a daemon that is gone names nothing
196pub async fn daemon_endpoint() -> Option<Endpoint> {
197    let body = tokio::fs::read_to_string(port_path().ok()?).await.ok()?;
198    let mut parts = body.split_whitespace();
199    let port: u16 = parts.next()?.parse().ok()?;
200    let token = parts.next()?.to_string();
201    let pid: u32 = parts.next()?.parse().ok()?;
202    let alive = tokio::task::spawn_blocking(move || crate::home::pid_is_orch_serve(pid))
203        .await
204        .ok()?;
205    alive.then(|| Endpoint {
206        url: format!("http://127.0.0.1:{port}/write"),
207        token,
208    })
209}
210
211pub(crate) async fn send(endpoint: &Endpoint, cmd: &Command) -> reqwest::Result<Reply> {
212    CLIENT
213        .post(&endpoint.url)
214        .bearer_auth(&endpoint.token)
215        .json(cmd)
216        .send()
217        .await?
218        .error_for_status()?
219        .json()
220        .await
221}
orch/cluster/local.rs

orch/cluster/
local.rs
