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 secret = Arc::new(cfg.secret);
268    let raft_store = RaftStore::open(raft_db, store.clone()).await?;
269    let (log, sm) = Adaptor::new(raft_store);
270    let config = openraft::Config {
271        enable_elect: cfg.elect,
272        ..config()?
273    }
274    .validate()?;
275    let network = network::HttpNetwork {
276        secret: secret.clone(),
277    };
278    let raft = Raft::new(cfg.id, Arc::new(config), network, log, sm).await?;
279    let listener = tokio::net::TcpListener::bind(&cfg.addr).await?;
280    let me = network::Hello {
281        name: cfg.name,
282        id: cfg.id,
283        witness: !cfg.elect,
284        leader: None,
285        term: 0,
286        proof: String::new(),
287    };
288    let app = network::router(raft.clone(), secret.clone(), me, cfg.addr.clone());
289    let stop = Arc::new(tokio::sync::Notify::new());
290    let stopped = stop.clone();
291    tokio::spawn(async move {
292        // Graceful shutdown also closes kept-alive connections, which would keep answering for a stopped node
293        let served = axum::serve(listener, app)
294            .with_graceful_shutdown(async move { stopped.notified().await })
295            .await;
296        if let Err(e) = served {
297            tracing::warn!(err = %e, "cluster listener stopped");
298        }
299    });
300    // initialize starts an election whatever enable_elect says, so a witness waits for the leader
301    if cfg.elect && !cfg.voters.is_empty() {
302        let members: BTreeMap<NodeId, BasicNode> = cfg
303            .voters
304            .into_iter()
305            .map(|(id, addr)| (id, BasicNode::new(addr)))
306            .collect();
307        if let Err(e) = raft.initialize(members).await {
308            tracing::debug!(err = %e, "raft already initialized");
309        }
310    }
311
312    let node = Node {
313        raft,
314        secret: Some(secret),
315        server: Some(stop),
316    };
317    store.attach_node(node.clone());
318    Ok(node)
319}
320
321/// A single node never sends an RPC. The tailnet transport replaces this
322struct NoNetwork;
323
324impl RaftNetworkFactory<TypeConfig> for NoNetwork {
325    type Network = NoNetwork;
326
327    async fn new_client(&mut self, _target: NodeId, _node: &BasicNode) -> NoNetwork {
328        NoNetwork
329    }
330}
331
332fn unreachable<E: std::error::Error>() -> RPCError<NodeId, BasicNode, E> {
333    RPCError::Unreachable(Unreachable::new(&std::io::Error::other(
334        "no network on a single node",
335    )))
336}
337
338impl RaftNetwork<TypeConfig> for NoNetwork {
339    async fn append_entries(
340        &mut self,
341        _req: AppendEntriesRequest<TypeConfig>,
342        _option: RPCOption,
343    ) -> Result<AppendEntriesResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
344        Err(unreachable())
345    }
346
347    async fn install_snapshot(
348        &mut self,
349        _req: InstallSnapshotRequest<TypeConfig>,
350        _option: RPCOption,
351    ) -> Result<
352        InstallSnapshotResponse<NodeId>,
353        RPCError<NodeId, BasicNode, RaftError<NodeId, InstallSnapshotError>>,
354    > {
355        Err(unreachable())
356    }
357
358    async fn vote(
359        &mut self,
360        _req: VoteRequest<NodeId>,
361        _option: RPCOption,
362    ) -> Result<VoteResponse<NodeId>, RPCError<NodeId, BasicNode, RaftError<NodeId>>> {
363        Err(unreachable())
364    }
365}
366
367#[cfg(test)]
368mod tests;
orch/cluster/mod.rs

orch/cluster/
mod.rs
