zuka
zuka/src/ci/run.rs

worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git
branch1be1d51cceb0018e6cc00721e0808a47f90f2f49 ▾
main All branches →
zuka/src/ci/run.rs
RSrun.rs11.6 KBDownload
1// The run record and where it lives on disk.
2//
3// One JSON file per run plus one log file, under
4// `$DATA_DIR/runs/{account}/{repo}/`. The same store backs the queue: a run written
5// as `Queued` is the queue entry, so a push that lands while the server is down is
6// still picked up when it comes back, with no separate durable channel to keep
7// consistent.
8
9use crate::error::{Error, Result};
10use crate::git::validate::Name;
11use anyhow::Context;
12use serde::{Deserialize, Serialize};
13use std::path::{Path, PathBuf};
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum Status {
18 Queued,
19 Running,
20 Succeeded,
21 Failed,
22 Cancelled,
23 TimedOut,
24}
25
26impl Status {
27 /// Whether the run is finished and will not change again.
28 pub fn terminal(self) -> bool {
29 !matches!(self, Status::Queued | Status::Running)
30 }
31}
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct Run {
35 pub id: String,
36 pub account: String,
37 pub repo: String,
38 /// The ref that triggered this run.
39 pub git_ref: String,
40 pub sha: String,
41 pub status: Status,
42 pub created_at: u64,
43 #[serde(default, skip_serializing_if = "Option::is_none")]
44 pub started_at: Option<u64>,
45 #[serde(default, skip_serializing_if = "Option::is_none")]
46 pub finished_at: Option<u64>,
47 /// Exit code of the step that ended the run.
48 #[serde(default, skip_serializing_if = "Option::is_none")]
49 pub exit_code: Option<i32>,
50 /// Why the run ended, when that is not obvious from the status.
51 #[serde(default, skip_serializing_if = "Option::is_none")]
52 pub detail: Option<String>,
53}
54
55impl Run {
56 pub fn queued(account: &str, repo: &str, git_ref: &str, sha: &str) -> Self {
57 Run {
58 id: uuid::Uuid::new_v4().simple().to_string(),
59 account: account.to_string(),
60 repo: repo.to_string(),
61 git_ref: git_ref.to_string(),
62 sha: sha.to_string(),
63 status: Status::Queued,
64 created_at: crate::account::token::now_secs(),
65 started_at: None,
66 finished_at: None,
67 exit_code: None,
68 detail: None,
69 }
70 }
71}
72
73/// Run records and their logs.
74#[derive(Debug, Clone)]
75pub struct RunStore {
76 root: PathBuf,
77}
78
79impl RunStore {
80 pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
81 let root = root.into();
82 std::fs::create_dir_all(&root)
83 .with_context(|| format!("create runs dir {}", root.display()))?;
84 Ok(RunStore { root })
85 }
86
87 fn dir(&self, account: &Name, repo: &Name) -> PathBuf {
88 self.root.join(account.as_str()).join(repo.as_str())
89 }
90
91 /// Run ids are generated, never supplied, but this is the boundary where a
92 /// client-supplied one would arrive — so it is checked before it becomes a path.
93 fn record_path(&self, account: &Name, repo: &Name, id: &str) -> Result<PathBuf> {
94 if id.is_empty() || id.len() > 64 || !id.bytes().all(|b| b.is_ascii_alphanumeric()) {
95 return Err(Error::invalid("invalid-run", "run id must be alphanumeric"));
96 }
97 Ok(self.dir(account, repo).join(format!("{id}.json")))
98 }
99
100 pub fn log_path(&self, account: &Name, repo: &Name, id: &str) -> Result<PathBuf> {
101 Ok(self.record_path(account, repo, id)?.with_extension("log"))
102 }
103
104 pub fn put(&self, account: &Name, repo: &Name, run: &Run) -> Result<()> {
105 let path = self.record_path(account, repo, &run.id)?;
106 let dir = path.parent().expect("record path has a parent");
107 std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
108
109 let body = serde_json::to_vec_pretty(run).context("serialize run")?;
110 let temp = path.with_extension("json.tmp");
111 std::fs::write(&temp, &body).with_context(|| format!("write {}", temp.display()))?;
112 std::fs::rename(&temp, &path).with_context(|| format!("rename into {}", path.display()))?;
113 Ok(())
114 }
115
116 pub fn get(&self, account: &Name, repo: &Name, id: &str) -> Result<Option<Run>> {
117 let path = self.record_path(account, repo, id)?;
118 match std::fs::read(&path) {
119 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
120 Err(e) => Err(Error::Internal(
121 anyhow::Error::from(e).context(format!("read {}", path.display())),
122 )),
123 Ok(bytes) => Ok(Some(
124 serde_json::from_slice(&bytes)
125 .with_context(|| format!("parse {}", path.display()))?,
126 )),
127 }
128 }
129
130 /// Runs for one repository, newest first.
131 pub fn list(&self, account: &Name, repo: &Name) -> Result<Vec<Run>> {
132 let mut runs = read_dir_runs(&self.dir(account, repo));
133 runs.sort_by(|a, b| b.created_at.cmp(&a.created_at).then(b.id.cmp(&a.id)));
134 Ok(runs)
135 }
136
137 /// Every queued run across every repository, oldest first.
138 ///
139 /// This is the queue. It is derived from the same records the API serves, so
140 /// there is no second source of truth to drift.
141 pub fn queued(&self) -> Vec<Run> {
142 let mut queued = Vec::new();
143 let Ok(accounts) = std::fs::read_dir(&self.root) else {
144 return queued;
145 };
146
147 for account in accounts.flatten() {
148 let Ok(repos) = std::fs::read_dir(account.path()) else {
149 continue;
150 };
151 for repo in repos.flatten() {
152 queued.extend(
153 read_dir_runs(&repo.path())
154 .into_iter()
155 .filter(|r| r.status == Status::Queued),
156 );
157 }
158 }
159 queued.sort_by(|a, b| a.created_at.cmp(&b.created_at).then(a.id.cmp(&b.id)));
160 queued
161 }
162
163 /// Append to a run's log, bounded.
164 ///
165 /// Returns how many bytes were actually written, so the caller can notice the
166 /// cap was reached and stop producing output.
167 pub fn append_log(
168 &self,
169 account: &Name,
170 repo: &Name,
171 id: &str,
172 chunk: &[u8],
173 cap: u64,
174 ) -> Result<usize> {
175 use std::io::Write;
176
177 let path = self.log_path(account, repo, id)?;
178 let dir = path.parent().expect("log path has a parent");
179 std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
180
181 let existing = std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
182 if existing >= cap {
183 return Ok(0);
184 }
185 let room = (cap - existing) as usize;
186 let slice = &chunk[..chunk.len().min(room)];
187
188 let mut file = std::fs::OpenOptions::new()
189 .create(true)
190 .append(true)
191 .open(&path)
192 .with_context(|| format!("open {}", path.display()))?;
193 file.write_all(slice)
194 .with_context(|| format!("write {}", path.display()))?;
195 Ok(slice.len())
196 }
197
198 pub fn read_log(&self, account: &Name, repo: &Name, id: &str) -> Result<Vec<u8>> {
199 let path = self.log_path(account, repo, id)?;
200 match std::fs::read(&path) {
201 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
202 Err(e) => Err(Error::Internal(
203 anyhow::Error::from(e).context(format!("read {}", path.display())),
204 )),
205 Ok(bytes) => Ok(bytes),
206 }
207 }
208
209 /// Remove every run for a repository. Called when the repository is deleted.
210 pub fn remove_repo(&self, account: &Name, repo: &Name) {
211 let _ = std::fs::remove_dir_all(self.dir(account, repo));
212 }
213}
214
215fn read_dir_runs(dir: &Path) -> Vec<Run> {
216 let Ok(entries) = std::fs::read_dir(dir) else {
217 return Vec::new();
218 };
219 entries
220 .flatten()
221 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
222 .filter_map(|e| std::fs::read(e.path()).ok())
223 .filter_map(|bytes| serde_json::from_slice::<Run>(&bytes).ok())
224 .collect()
225}
226
227#[cfg(test)]
228mod tests {
229 use super::*;
230 use crate::git::validate::name;
231
232 fn store() -> (tempfile::TempDir, RunStore, Name, Name) {
233 let dir = tempfile::tempdir().unwrap();
234 let store = RunStore::open(dir.path()).unwrap();
235 (dir, store, name("alice").unwrap(), name("site").unwrap())
236 }
237
238 #[test]
239 fn a_run_round_trips() {
240 let (_dir, store, a, r) = store();
241 let run = Run::queued("alice", "site", "refs/heads/main", "abc");
242 store.put(&a, &r, &run).unwrap();
243
244 let found = store.get(&a, &r, &run.id).unwrap().unwrap();
245 assert_eq!(found.status, Status::Queued);
246 assert_eq!(found.git_ref, "refs/heads/main");
247 }
248
249 #[test]
250 fn terminal_states_are_the_ones_that_stop_changing() {
251 assert!(!Status::Queued.terminal());
252 assert!(!Status::Running.terminal());
253 for status in [
254 Status::Succeeded,
255 Status::Failed,
256 Status::Cancelled,
257 Status::TimedOut,
258 ] {
259 assert!(status.terminal(), "{status:?} must be terminal");
260 }
261 }
262
263 #[test]
264 fn the_queue_is_derived_from_the_records_the_api_serves() {
265 let (_dir, store, a, r) = store();
266
267 let mut first = Run::queued("alice", "site", "refs/heads/main", "a");
268 first.created_at = 100;
269 let mut second = Run::queued("alice", "site", "refs/heads/main", "b");
270 second.created_at = 200;
271 let mut done = Run::queued("alice", "site", "refs/heads/main", "c");
272 done.created_at = 50;
273 done.status = Status::Succeeded;
274
275 for run in [&first, &second, &done] {
276 store.put(&a, &r, run).unwrap();
277 }
278
279 let queued = store.queued();
280 assert_eq!(queued.len(), 2, "a finished run must not be in the queue");
281 assert_eq!(queued[0].id, first.id, "the queue is oldest first");
282 assert_eq!(queued[1].id, second.id);
283 }
284
285 #[test]
286 fn listing_is_newest_first() {
287 let (_dir, store, a, r) = store();
288 let mut old = Run::queued("alice", "site", "refs/heads/main", "a");
289 old.created_at = 100;
290 let mut new = Run::queued("alice", "site", "refs/heads/main", "b");
291 new.created_at = 200;
292 store.put(&a, &r, &old).unwrap();
293 store.put(&a, &r, &new).unwrap();
294
295 let listed = store.list(&a, &r).unwrap();
296 assert_eq!(listed[0].id, new.id);
297 }
298
299 #[test]
300 fn a_log_is_capped_and_reports_what_it_wrote() {
301 let (_dir, store, a, r) = store();
302 let run = Run::queued("alice", "site", "refs/heads/main", "a");
303
304 assert_eq!(store.append_log(&a, &r, &run.id, b"12345", 8).unwrap(), 5);
305 // Only three bytes of room are left.
306 assert_eq!(store.append_log(&a, &r, &run.id, b"67890", 8).unwrap(), 3);
307 // And then none.
308 assert_eq!(store.append_log(&a, &r, &run.id, b"more", 8).unwrap(), 0);
309
310 assert_eq!(store.read_log(&a, &r, &run.id).unwrap(), b"12345678");
311 }
312
313 #[test]
314 fn an_absent_log_reads_as_empty_rather_than_failing() {
315 let (_dir, store, a, r) = store();
316 assert!(store.read_log(&a, &r, "deadbeef").unwrap().is_empty());
317 }
318
319 #[test]
320 fn a_run_id_cannot_become_a_path() {
321 let (_dir, store, a, r) = store();
322 for id in ["../../etc/passwd", "a/b", "..", "", "a.json"] {
323 assert!(store.get(&a, &r, id).is_err(), "{id:?} must be refused");
324 }
325 }
326
327 #[test]
328 fn removing_a_repository_removes_its_runs() {
329 let (_dir, store, a, r) = store();
330 let run = Run::queued("alice", "site", "refs/heads/main", "a");
331 store.put(&a, &r, &run).unwrap();
332
333 store.remove_repo(&a, &r);
334 assert!(store.get(&a, &r, &run.id).unwrap().is_none());
335 assert!(store.queued().is_empty());
336 }
337}