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

orch/cluster/
command.rs
