orch/cluster/
command.rs

1use serde::{Deserialize, Serialize};
2use sqlx::SqliteConnection;
3
4use crate::store::control::MemoryEntry;
5use crate::store::dependencies::{self, NewDependency};
6use crate::store::jobs::{self, JobState};
7use crate::store::kv;
8use crate::store::tasks;
9use crate::store::tasks::{NewTask, StoreRouteCmd, TaskStatus, UpsertExternalCmd, apply};
10
11type Updates = Vec<(String, serde_json::Value)>;
12
13/// One replicated write. The proposer stamps `now`, so every node applies the same value
14#[derive(Serialize, Deserialize, Clone, Debug)]
15pub struct Request {
16    pub now: String,
17    pub cmd: Command,
18}
19
20#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
21#[serde(tag = "op", rename_all = "snake_case")]
22pub enum Command {
23    KvSet {
24        key: String,
25        value: String,
26    },
27    KvIncrement {
28        key: String,
29    },
30    KvInsertIfAbsent {
31        key: String,
32        value: String,
33    },
34    KvDelete {
35        key: String,
36    },
37    TaskCreate {
38        new: NewTask,
39        internal: bool,
40    },
41    TaskUpsertExternal(UpsertExternalCmd),
42    TaskUpdateStatusAndFields {
43        id: i64,
44        status: TaskStatus,
45        updates: Updates,
46    },
47    TaskUpdateStatus {
48        id: i64,
49        status: TaskStatus,
50        expected: Option<TaskStatus>,
51    },
52    TaskUpdateExternalId {
53        id: i64,
54        external_id: String,
55    },
56    TaskTouch {
57        id: i64,
58    },
59    TaskResetToNew {
60        id: i64,
61    },
62    TaskSetFields {
63        id: i64,
64        updates: Updates,
65        silent: bool,
66    },
67    TaskIncrement {
68        id: i64,
69        field: String,
70    },
71    TaskResetCounters {
72        id: i64,
73        failure_only: bool,
74    },
75    TaskStoreRoute(StoreRouteCmd),
76    TaskSetEstimateIfZero {
77        id: i64,
78        estimate: u8,
79    },
80    TaskAppendMemory {
81        id: i64,
82        entry: MemoryEntry,
83    },
84    TaskMarkCleaned {
85        id: i64,
86    },
87    MoveRepo {
88        old: String,
89        new: String,
90    },
91    DepReplace {
92        task_id: i64,
93        rows: Vec<NewDependency>,
94    },
95    DepLinkKnown {
96        task_id: i64,
97        repo: String,
98        number: i64,
99    },
100    DepBlockClaim {
101        task_id: i64,
102    },
103    JobStateUpsert(JobState),
104    ChannelSubscribe {
105        channel: String,
106        thread_id: String,
107        repo: String,
108        topic_id: String,
109    },
110    ChannelUnsubscribe {
111        channel: String,
112        thread_id: String,
113        repo: String,
114    },
115}
116
117/// A failed command is `Err` on every node, so one bad command cannot stop the state machine
118#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
119pub enum Reply {
120    Done,
121    Bool(bool),
122    Int(i64),
123    Count(u64),
124    Value(String),
125    Unmet(Option<Vec<String>>),
126    Err(String),
127}
128
129impl Command {
130    /// A write through the log may not touch this node's own state: a local kv key, such as its
131    /// Raft position, or a local task column, such as the worktree path that cleanup removes
132    pub(crate) fn check_replicable(&self) -> anyhow::Result<()> {
133        let key = match self {
134            Command::KvSet { key, .. }
135            | Command::KvIncrement { key }
136            | Command::KvInsertIfAbsent { key, .. }
137            | Command::KvDelete { key } => Some(key),
138            _ => None,
139        };
140        if let Some(key) = key {
141            anyhow::ensure!(
142                !kv::is_local(key),
143                "{key} is local to each node, never replicated"
144            );
145        }
146
147        let updates = match self {
148            Command::TaskSetFields { updates, .. }
149            | Command::TaskUpdateStatusAndFields { updates, .. } => &updates[..],
150            _ => &[],
151        };
152        if let Some((col, _)) = updates.iter().find(|(col, _)| tasks::is_local_column(col)) {
153            anyhow::bail!("{col} is local to each node, never replicated");
154        }
155
156        Ok(())
157    }
158
159    /// `replicated` is true when Raft applies the command, so new ids come from the replicated
160    /// rows. A direct write keeps SQLite's AUTOINCREMENT, as orch on one machine always has
161    pub(crate) async fn apply(
162        self,
163        conn: &mut SqliteConnection,
164        now: &str,
165        replicated: bool,
166    ) -> anyhow::Result<Reply> {
167        if replicated {
168            self.check_replicable()?;
169        }
170
171        Ok(match self {
172            Command::KvSet { key, value } => {
173                kv::set(conn, now, &key, &value).await?;
174                Reply::Done
175            }
176            Command::KvIncrement { key } => Reply::Count(kv::increment(conn, now, &key).await?),
177            Command::KvInsertIfAbsent { key, value } => {
178                Reply::Value(kv::insert_if_absent(conn, now, &key, &value).await?)
179            }
180            Command::KvDelete { key } => {
181                kv::delete(conn, &key).await?;
182                Reply::Done
183            }
184            Command::TaskCreate { new, internal } => {
185                Reply::Int(apply::create(conn, now, &new, internal, replicated).await?)
186            }
187            Command::TaskUpsertExternal(ext) => {
188                Reply::Int(apply::upsert_external(conn, now, &ext, replicated).await?)
189            }
190            Command::TaskUpdateStatusAndFields {
191                id,
192                status,
193                updates,
194            } => {
195                apply::update_status_and_fields(conn, now, id, status, &updates).await?;
196                Reply::Done
197            }
198            Command::TaskUpdateStatus {
199                id,
200                status,
201                expected,
202            } => Reply::Bool(apply::update_status(conn, now, id, status, expected).await?),
203            Command::TaskUpdateExternalId { id, external_id } => {
204                apply::update_external_id(conn, now, id, &external_id).await?;
205                Reply::Done
206            }
207            Command::TaskTouch { id } => {
208                apply::touch(conn, now, id).await?;
209                Reply::Done
210            }
211            Command::TaskResetToNew { id } => {
212                apply::reset_to_new(conn, now, id).await?;
213                Reply::Done
214            }
215            Command::TaskSetFields {
216                id,
217                updates,
218                silent,
219            } => {
220                apply::set_fields(conn, now, id, &updates, silent).await?;
221                Reply::Done
222            }
223            Command::TaskIncrement { id, field } => {
224                Reply::Int(apply::increment(conn, now, id, &field).await?)
225            }
226            Command::TaskResetCounters { id, failure_only } => {
227                apply::reset_counters(conn, now, id, failure_only).await?;
228                Reply::Done
229            }
230            Command::TaskStoreRoute(route) => {
231                apply::store_route(conn, now, &route).await?;
232                Reply::Done
233            }
234            Command::TaskSetEstimateIfZero { id, estimate } => {
235                Reply::Bool(apply::set_estimate_if_zero(conn, now, id, estimate).await?)
236            }
237            Command::TaskAppendMemory { id, entry } => {
238                apply::append_memory(conn, now, id, &entry).await?;
239                Reply::Done
240            }
241            Command::TaskMarkCleaned { id } => {
242                apply::mark_cleaned(conn, now, id).await?;
243                Reply::Done
244            }
245            Command::MoveRepo { old, new } => {
246                let moved = dependencies::apply::move_repo(conn, &old, &new).await?;
247                Reply::Int(i64::try_from(moved)?)
248            }
249            Command::DepReplace { task_id, rows } => {
250                dependencies::apply::replace_dependencies(conn, task_id, &rows).await?;
251                Reply::Done
252            }
253            Command::DepLinkKnown {
254                task_id,
255                repo,
256                number,
257            } => {
258                dependencies::apply::link_known_task(conn, task_id, &repo, number).await?;
259                Reply::Done
260            }
261            Command::DepBlockClaim { task_id } => {
262                Reply::Unmet(dependencies::apply::block_claim(conn, now, task_id).await?)
263            }
264            Command::JobStateUpsert(state) => {
265                jobs::apply::upsert_job_state(conn, &state).await?;
266                Reply::Done
267            }
268            Command::ChannelSubscribe {
269                channel,
270                thread_id,
271                repo,
272                topic_id,
273            } => {
274                jobs::apply::subscribe(conn, &channel, &thread_id, &repo, &topic_id).await?;
275                Reply::Done
276            }
277            Command::ChannelUnsubscribe {
278                channel,
279                thread_id,
280                repo,
281            } => {
282                jobs::apply::unsubscribe(conn, &channel, &thread_id, &repo).await?;
283                Reply::Done
284            }
285        })
286    }
287}
orch/cluster/command.rs

orch/cluster/
command.rs
