orch/cluster/
mod.rs

1//! Raft between orch nodes. Only coordination state goes through the log, the design is in
2//! `docs/specs/2026-10-26-multi-machine-cluster-design.md`.
3
4mod command;
5pub mod config;
6pub mod discovery;
7pub mod local;
8pub mod network;
9pub mod progress;
10mod storage;
11
12use std::collections::BTreeMap;
13use std::io::Cursor;
14use std::path::Path;
15use std::sync::Arc;
16use std::time::Duration;
17
18use openraft::error::{ClientWriteError, InstallSnapshotError, RPCError, RaftError, Unreachable};
19use openraft::network::{RPCOption, RaftNetwork, RaftNetworkFactory};
20use openraft::raft::{
21    AppendEntriesRequest, AppendEntriesResponse, InstallSnapshotRequest, InstallSnapshotResponse,
22    VoteRequest, VoteResponse,
23};
24use openraft::storage::Adaptor;
25use openraft::{BasicNode, SnapshotPolicy};
26
27use crate::store::TaskStore;
28
29pub use command::{Command, Reply, Request};
30pub use storage::RaftStore;
31
32pub type NodeId = u64;
33
34openraft::declare_raft_types!(
35    pub TypeConfig:
36        D = Request,
37        R = Reply,
38);
39
40pub type Raft = openraft::Raft<TypeConfig>;
41
42/// A `cluster:` section turns the cluster on. Without one, orch writes straight to SQLite as it always has
43pub fn enabled() -> bool {
44    crate::config::get("cluster.secret").is_ok_and(|s| !s.is_empty())
45}
46
47/// Start this machine's node from the `cluster:` section and the tailnet. A learner keeps asking
48/// the cluster to add it until a member proves the secret and takes it
49pub async fn start(store: &TaskStore, raft_db: &Path) -> anyhow::Result<Node> {
50    let cfg = config::ClusterConfig::load()?
51        .ok_or_else(|| anyhow::anyhow!("no cluster.secret in the config"))?;
52    let tailnet = discovery::status().await?;
53    let plan = config::plan(&cfg, &tailnet)?;
54    tracing::info!(
55        name = %plan.name,
56        addr = %plan.addr,
57        elect = plan.elect,
58        learner = plan.learner,
59        "starting cluster node"
60    );
61    let node = start_node(
62        store,
63        raft_db,
64        NodeConfig {
65            name: plan.name.clone(),
66            id: plan.id,
67            addr: plan.addr.clone(),
68            secret: cfg.secret.clone(),
69            voters: plan.voters.clone(),
70            elect: plan.elect,
71        },
72    )
73    .await?;
74    if plan.learner {
75        tokio::spawn(join_loop(node.clone(), plan, cfg));
76    }
77
78    Ok(node)
79}
80
81async fn join_loop(node: Node, plan: config::Plan, cfg: config::ClusterConfig) {
82    loop {
83        let member = node
84            .raft
85            .metrics()
86            .borrow()
87            .membership_config
88            .membership()
89            .nodes()
90            .any(|(id, _)| *id == plan.id);
91        if member {
92            return;
93        }
94
95        if let Ok(tailnet) = discovery::status().await {
96            for peer in tailnet.peers.iter().filter(|p| p.online) {
97                let addr = format!("{}:{}", peer.ip, cfg.port);
98                if network::probe(&addr, &cfg.secret, Some(&peer.name))
99                    .await
100                    .is_none()
101                {
102                    continue;
103                }
104
105                let join = network::Join {
106                    id: plan.id,
107                    addr: plan.addr.clone(),
108                };
109                match network::send_join(&addr, &cfg.secret, &join).await {
110                    Ok(()) => tracing::info!(via = %peer.name, "joined the cluster as a learner"),
111                    Err(e) => tracing::warn!(via = %peer.name, err = %e, "join refused"),
112                }
113
114                break;
115            }
116        }
117
118        tokio::time::sleep(Duration::from_secs(30)).await;
119    }
120}
121
122/// One member as the leader sees it, for `orch nodes`
123#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
124pub struct Member {
125    pub id: NodeId,
126    pub addr: String,
127    pub voter: bool,
128    pub leader: bool,
129    pub me: bool,
130}
131
132pub fn members(node: &Node) -> Vec<Member> {
133    let m = node.raft.metrics().borrow().clone();
134    let membership = m.membership_config.membership();
135    let voters: std::collections::BTreeSet<NodeId> = membership.voter_ids().collect();
136    membership
137        .nodes()
138        .map(|(id, n)| Member {
139            id: *id,
140            addr: n.addr.clone(),
141            voter: voters.contains(id),
142            leader: m.current_leader == Some(*id),
143            me: *id == m.id,
144        })
145        .collect()
146}
147
148/// Id of the node when orch runs alone
149pub const SINGLE_NODE_ID: NodeId = 1;
150
151/// The proposer's clock, in the format the store writes timestamps in
152pub fn now_stamp() -> String {
153    chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()
154}
155
156fn config() -> anyhow::Result<openraft::Config> {
157    Ok(openraft::Config {
158        cluster_name: "orch".into(),
159        heartbeat_interval: 250,
160        election_timeout_min: 1000,
161        election_timeout_max: 2000,
162        snapshot_policy: SnapshotPolicy::LogsSinceLast(1000),
163        ..Default::default()
164    }
165    .validate()?)
166}
167
168/// A running node. The store holds one and sends every replicated write through it
169#[derive(Clone)]
170pub struct Node {
171    pub raft: Raft,
172    /// None on a single node, which has no peers to forward to
173    secret: Option<Arc<String>>,
174    server: Option<Arc<tokio::sync::Notify>>,
175}
176
177impl Node {
178    /// Propose on this node, or forward to the leader when this node follows. During an election
179    /// the leader can be unknown or stale, so it waits and tries again, but only a write that was
180    /// certainly not applied is sent again
181    pub async fn write(&self, req: Request) -> anyhow::Result<Reply> {
182        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
183        loop {
184            let err = match self.raft.client_write(req.clone()).await {
185                Ok(r) => return Ok(r.data),
186                Err(e) => e,
187            };
188            let RaftError::APIError(ClientWriteError::ForwardToLeader(f)) = &err else {
189                return Err(err.into());
190            };
191            let Some(secret) = &self.secret else {
192                return Err(err.into());
193            };
194            let reason = match &f.leader_node {
195                Some(leader) => match network::forward(&leader.addr, secret, &req).await {
196                    Ok(reply) => return Ok(reply),
197                    Err(network::ForwardError::Unknown(e)) => return Err(e),
198                    Err(network::ForwardError::NotApplied(why)) => why,
199                },
200                None => "no leader known".to_string(),
201            };
202            if tokio::time::Instant::now() >= deadline {
203                anyhow::bail!("no leader took the write: {reason}");
204            }
205
206            tokio::time::sleep(Duration::from_millis(200)).await;
207        }
208    }
209
210    pub fn is_leader(&self) -> bool {
211        let m = self.raft.metrics();
212        let m = m.borrow();
213        m.current_leader == Some(m.id)
214    }
215
216    pub async fn stop(self) {
217        if let Some(server) = self.server {
218            server.notify_one();
219        }
220
221        let _ = self.raft.shutdown().await;
222    }
223}
224
225/// How one node of a cluster starts: its id, the address it listens on, the shared secret, the
226/// voters to initialize with, and whether it may lead (a witness may not)
227pub struct NodeConfig {
228    pub name: String,
229    pub id: NodeId,
230    pub addr: String,
231    pub secret: String,
232    pub voters: BTreeMap<NodeId, String>,
233    pub elect: bool,
234}
235
236/// Start a one-node cluster over `store`, with its log in `raft_db`, and route the store's
237/// replicated writes through it
238pub async fn start_single(store: &TaskStore, raft_db: &Path) -> anyhow::Result<Node> {
239    let raft_store = RaftStore::open(raft_db, store.clone()).await?;
240    let (log, sm) = Adaptor::new(raft_store);
241    let raft = Raft::new(SINGLE_NODE_ID, Arc::new(config()?), NoNetwork, log, sm).await?;
242    let members = BTreeMap::from([(SINGLE_NODE_ID, BasicNode::default())]);
243    // A restart finds the membership in the log, and initialize refuses a second time
244    if let Err(e) = raft.initialize(members).await {
245        tracing::debug!(err = %e, "raft already initialized");
246    }
247
248    raft.wait(Some(Duration::from_secs(10)))
249        .current_leader(SINGLE_NODE_ID, "single node leads")
250        .await?;
251    let node = Node {
252        raft,
253        secret: None,
254        server: None,
255    };
256    store.attach_node(node.clone());
257    Ok(node)
258}
259
260/// Start one node of a cluster and serve its RPCs on `cfg.addr`. Every voter initializes with
261/// the same membership, which openraft accepts, so no voter is special
262pub async fn start_node(
263    store: &TaskStore,
264    raft_db: &Path,
265    cfg: NodeConfig,
266) -> anyhow::Result<Node> {
267    let listener = tokio::net::TcpListener::bind(&cfg.addr).await?;
268    start_node_on(store, raft_db, cfg, listener).await
269}
270
271/// Like `start_node`, with a listener the caller already bound. `cfg.addr` is the address peers use
272pub async fn start_node_on(
273    store: &TaskStore,
274    raft_db: &Path,
275    cfg: NodeConfig,
276    listener: tokio::net::TcpListener,
277) -> anyhow::Result<Node> {
278    let secret = Arc::new(cfg.secret);
279    let raft_store = RaftStore::open(raft_db, store.clone()).await?;
280    let (log, sm) = Adaptor::new(raft_store);
281    let config = openraft::Config {
282        enable_elect: cfg.elect,
283        ..config()?
284    }
285    .validate()?;
286    let network = network::HttpNetwork {
287        secret: secret.clone(),
288    };
289    let raft = Raft::new(cfg.id, Arc::new(config), network, log, sm).await?;
290    let me = network::Hello {
291        name: cfg.name,
292        id: cfg.id,
293        witness: !cfg.elect,
294        leader: None,
295        term: 0,
296        proof: String::new(),
297    };
298    let app = network::router(raft.clone(), secret.clone(), me, cfg.addr.clone());
299    let stop = Arc::new(tokio::sync::Notify::new());
300    let stopped = stop.clone();
301    tokio::spawn(async move {
302        // Graceful shutdown also closes kept-alive connections, which would keep answering for a stopped node
303        let served = axum::serve(listener, app)
304            .with_graceful_shutdown(async move { stopped.notified().await })
305            .await;
306        if let Err(e) = served {
307            tracing::warn!(err = %e, "cluster listener stopped");
308        }
309    });
310    // initialize starts an election whatever enable_elect says, so a witness waits for the leader
311    if cfg.elect && !cfg.voters.is_empty() {
312        let members: BTreeMap<NodeId, BasicNode> = cfg
313            .voters
314            .into_iter()
315            .map(|(id, addr)| (id, BasicNode::new(addr)))
316            .collect();
317        if let Err(e) = raft.initialize(members).await {
318            tracing::debug!(err = %e, "raft already initialized");
319        }
320    }
321
322    let node = Node {
323        raft,
324        secret: Some(secret),
325        server: Some(stop),
326    };
327    store.attach_node(node.clone());
328    Ok(node)
329}
330
331/// A single node never sends an RPC. The tailnet transport replaces this
332struct NoNetwork;
333
334impl RaftNetworkFactory<TypeConfig> for NoNetwork {
335    type Network = NoNetwork;
336
337    async fn new_client(&mut self, _target: NodeId, _node: &BasicNode) -> NoNetwork {
338        NoNetwork
339    }
340}
341
342fn unreachable<E: std::error::Error>() -> RPCError<NodeId, BasicNode, E> {
343    RPCError::Unreachable(Unreachable::new(&std::io::Error::other(
344        "no network on a single node",
345    )))
346}
347
348impl RaftNetwork<TypeConfig> for NoNetwork {
349    async fn append_entries(
350        &mut self,
351        _req: AppendEntriesRequest<TypeConfig>,
352        _option: RPCOption,
353    ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
354        Err(unreachable())
355    }
356
357    async fn install_snapshot(
358        &mut self,
359        _req: InstallSnapshotRequest<TypeConfig>,
360        _option: RPCOption,
361    ) -> Result<
362        InstallSnapshotResponse<NodeId>,
363        RPCError<NodeId, BasicNode, RaftError<NodeId, InstallSnapshotError>>,
364    > {
365        Err(unreachable())
366    }
367
368    async fn vote(
369        &mut self,
370        _req: VoteRequest<NodeId>,
371        _option: RPCOption,
372    ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
373        Err(unreachable())
374    }
375}
376
377#[cfg(test)]
378mod tests;
orch/cluster/mod.rs

orch/cluster/
mod.rs
