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
branch5cca6cb3cf778c2d00882389a591d4fc1cf0b21b ▾
main All branches →
zuka/src/ci/run.rs
RSrun.rs12.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 /// Position of this run in the repository's history, from 1.
43 ///
44 /// A build number, in the ordinary sense: something that grows and that a person
45 /// can compare at a glance. The run id is a uuid, which is unique but says
46 /// nothing about order. Records written before this existed report 0, which no
47 /// real run ever has.
48 #[serde(default)]
49 pub number: u64,
50 pub created_at: u64,
51 #[serde(default, skip_serializing_if = "Option::is_none")]
52 pub started_at: Option<u64>,
53 #[serde(default, skip_serializing_if = "Option::is_none")]
54 pub finished_at: Option<u64>,
55 /// Exit code of the step that ended the run.
56 #[serde(default, skip_serializing_if = "Option::is_none")]
57 pub exit_code: Option<i32>,
58 /// Why the run ended, when that is not obvious from the status.
59 #[serde(default, skip_serializing_if = "Option::is_none")]
60 pub detail: Option<String>,
61}
62
63impl Run {
64 pub fn queued(account: &str, repo: &str, git_ref: &str, sha: &str) -> Self {
65 Run {
66 id: uuid::Uuid::new_v4().simple().to_string(),
67 account: account.to_string(),
68 repo: repo.to_string(),
69 git_ref: git_ref.to_string(),
70 sha: sha.to_string(),
71 status: Status::Queued,
72 number: 0,
73 created_at: crate::account::token::now_secs(),
74 started_at: None,
75 finished_at: None,
76 exit_code: None,
77 detail: None,
78 }
79 }
80}
81
82/// Run records and their logs.
83#[derive(Debug, Clone)]
84pub struct RunStore {
85 root: PathBuf,
86}
87
88impl RunStore {
89 pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
90 let root = root.into();
91 std::fs::create_dir_all(&root)
92 .with_context(|| format!("create runs dir {}", root.display()))?;
93 Ok(RunStore { root })
94 }
95
96 fn dir(&self, account: &Name, repo: &Name) -> PathBuf {
97 self.root.join(account.as_str()).join(repo.as_str())
98 }
99
100 /// Run ids are generated, never supplied, but this is the boundary where a
101 /// client-supplied one would arrive — so it is checked before it becomes a path.
102 fn record_path(&self, account: &Name, repo: &Name, id: &str) -> Result<PathBuf> {
103 if id.is_empty() || id.len() > 64 || !id.bytes().all(|b| b.is_ascii_alphanumeric()) {
104 return Err(Error::invalid("invalid-run", "run id must be alphanumeric"));
105 }
106 Ok(self.dir(account, repo).join(format!("{id}.json")))
107 }
108
109 pub fn log_path(&self, account: &Name, repo: &Name, id: &str) -> Result<PathBuf> {
110 Ok(self.record_path(account, repo, id)?.with_extension("log"))
111 }
112
113 /// Take the next build number for a repository.
114 ///
115 /// Counts existing records rather than keeping a separate counter, because a
116 /// counter is a second thing to keep consistent with the records and would
117 /// disagree with them after any pruning. The consequence is that numbers can
118 /// repeat if old runs are pruned, which is worth the absence of a file that can
119 /// drift.
120 pub fn next_number(&self, account: &Name, repo: &Name) -> u64 {
121 self.list(account, repo)
122 .map(|runs| runs.iter().map(|r| r.number).max().unwrap_or(0) + 1)
123 .unwrap_or(1)
124 }
125
126 pub fn put(&self, account: &Name, repo: &Name, run: &Run) -> Result<()> {
127 let path = self.record_path(account, repo, &run.id)?;
128 let dir = path.parent().expect("record path has a parent");
129 std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
130
131 let body = serde_json::to_vec_pretty(run).context("serialize run")?;
132 let temp = path.with_extension("json.tmp");
133 std::fs::write(&temp, &body).with_context(|| format!("write {}", temp.display()))?;
134 std::fs::rename(&temp, &path).with_context(|| format!("rename into {}", path.display()))?;
135 Ok(())
136 }
137
138 pub fn get(&self, account: &Name, repo: &Name, id: &str) -> Result<Option<Run>> {
139 let path = self.record_path(account, repo, id)?;
140 match std::fs::read(&path) {
141 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
142 Err(e) => Err(Error::Internal(
143 anyhow::Error::from(e).context(format!("read {}", path.display())),
144 )),
145 Ok(bytes) => Ok(Some(
146 serde_json::from_slice(&bytes)
147 .with_context(|| format!("parse {}", path.display()))?,
148 )),
149 }
150 }
151
152 /// Runs for one repository, newest first.
153 pub fn list(&self, account: &Name, repo: &Name) -> Result<Vec<Run>> {
154 let mut runs = read_dir_runs(&self.dir(account, repo));
155 runs.sort_by(|a, b| b.created_at.cmp(&a.created_at).then(b.id.cmp(&a.id)));
156 Ok(runs)
157 }
158
159 /// Every queued run across every repository, oldest first.
160 ///
161 /// This is the queue. It is derived from the same records the API serves, so
162 /// there is no second source of truth to drift.
163 pub fn queued(&self) -> Vec<Run> {
164 let mut queued = Vec::new();
165 let Ok(accounts) = std::fs::read_dir(&self.root) else {
166 return queued;
167 };
168
169 for account in accounts.flatten() {
170 let Ok(repos) = std::fs::read_dir(account.path()) else {
171 continue;
172 };
173 for repo in repos.flatten() {
174 queued.extend(
175 read_dir_runs(&repo.path())
176 .into_iter()
177 .filter(|r| r.status == Status::Queued),
178 );
179 }
180 }
181 queued.sort_by(|a, b| a.created_at.cmp(&b.created_at).then(a.id.cmp(&b.id)));
182 queued
183 }
184
185 /// Append to a run's log, bounded.
186 ///
187 /// Returns how many bytes were actually written, so the caller can notice the
188 /// cap was reached and stop producing output.
189 pub fn append_log(
190 &self,
191 account: &Name,
192 repo: &Name,
193 id: &str,
194 chunk: &[u8],
195 cap: u64,
196 ) -> Result<usize> {
197 use std::io::Write;
198
199 let path = self.log_path(account, repo, id)?;
200 let dir = path.parent().expect("log path has a parent");
201 std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
202
203 let existing = std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0);
204 if existing >= cap {
205 return Ok(0);
206 }
207 let room = (cap - existing) as usize;
208 let slice = &chunk[..chunk.len().min(room)];
209
210 let mut file = std::fs::OpenOptions::new()
211 .create(true)
212 .append(true)
213 .open(&path)
214 .with_context(|| format!("open {}", path.display()))?;
215 file.write_all(slice)
216 .with_context(|| format!("write {}", path.display()))?;
217 Ok(slice.len())
218 }
219
220 pub fn read_log(&self, account: &Name, repo: &Name, id: &str) -> Result<Vec<u8>> {
221 let path = self.log_path(account, repo, id)?;
222 match std::fs::read(&path) {
223 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
224 Err(e) => Err(Error::Internal(
225 anyhow::Error::from(e).context(format!("read {}", path.display())),
226 )),
227 Ok(bytes) => Ok(bytes),
228 }
229 }
230
231 /// Remove every run for a repository. Called when the repository is deleted.
232 pub fn remove_repo(&self, account: &Name, repo: &Name) {
233 let _ = std::fs::remove_dir_all(self.dir(account, repo));
234 }
235}
236
237fn read_dir_runs(dir: &Path) -> Vec<Run> {
238 let Ok(entries) = std::fs::read_dir(dir) else {
239 return Vec::new();
240 };
241 entries
242 .flatten()
243 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("json"))
244 .filter_map(|e| std::fs::read(e.path()).ok())
245 .filter_map(|bytes| serde_json::from_slice::<Run>(&bytes).ok())
246 .collect()
247}
248
249#[cfg(test)]
250mod tests {
251 use super::*;
252 use crate::git::validate::name;
253
254 fn store() -> (tempfile::TempDir, RunStore, Name, Name) {
255 let dir = tempfile::tempdir().unwrap();
256 let store = RunStore::open(dir.path()).unwrap();
257 (dir, store, name("alice").unwrap(), name("site").unwrap())
258 }
259
260 #[test]
261 fn a_run_round_trips() {
262 let (_dir, store, a, r) = store();
263 let run = Run::queued("alice", "site", "refs/heads/main", "abc");
264 store.put(&a, &r, &run).unwrap();
265
266 let found = store.get(&a, &r, &run.id).unwrap().unwrap();
267 assert_eq!(found.status, Status::Queued);
268 assert_eq!(found.git_ref, "refs/heads/main");
269 }
270
271 #[test]
272 fn terminal_states_are_the_ones_that_stop_changing() {
273 assert!(!Status::Queued.terminal());
274 assert!(!Status::Running.terminal());
275 for status in [
276 Status::Succeeded,
277 Status::Failed,
278 Status::Cancelled,
279 Status::TimedOut,
280 ] {
281 assert!(status.terminal(), "{status:?} must be terminal");
282 }
283 }
284
285 #[test]
286 fn the_queue_is_derived_from_the_records_the_api_serves() {
287 let (_dir, store, a, r) = store();
288
289 let mut first = Run::queued("alice", "site", "refs/heads/main", "a");
290 first.created_at = 100;
291 let mut second = Run::queued("alice", "site", "refs/heads/main", "b");
292 second.created_at = 200;
293 let mut done = Run::queued("alice", "site", "refs/heads/main", "c");
294 done.created_at = 50;
295 done.status = Status::Succeeded;
296
297 for run in [&first, &second, &done] {
298 store.put(&a, &r, run).unwrap();
299 }
300
301 let queued = store.queued();
302 assert_eq!(queued.len(), 2, "a finished run must not be in the queue");
303 assert_eq!(queued[0].id, first.id, "the queue is oldest first");
304 assert_eq!(queued[1].id, second.id);
305 }
306
307 #[test]
308 fn listing_is_newest_first() {
309 let (_dir, store, a, r) = store();
310 let mut old = Run::queued("alice", "site", "refs/heads/main", "a");
311 old.created_at = 100;
312 let mut new = Run::queued("alice", "site", "refs/heads/main", "b");
313 new.created_at = 200;
314 store.put(&a, &r, &old).unwrap();
315 store.put(&a, &r, &new).unwrap();
316
317 let listed = store.list(&a, &r).unwrap();
318 assert_eq!(listed[0].id, new.id);
319 }
320
321 #[test]
322 fn a_log_is_capped_and_reports_what_it_wrote() {
323 let (_dir, store, a, r) = store();
324 let run = Run::queued("alice", "site", "refs/heads/main", "a");
325
326 assert_eq!(store.append_log(&a, &r, &run.id, b"12345", 8).unwrap(), 5);
327 // Only three bytes of room are left.
328 assert_eq!(store.append_log(&a, &r, &run.id, b"67890", 8).unwrap(), 3);
329 // And then none.
330 assert_eq!(store.append_log(&a, &r, &run.id, b"more", 8).unwrap(), 0);
331
332 assert_eq!(store.read_log(&a, &r, &run.id).unwrap(), b"12345678");
333 }
334
335 #[test]
336 fn an_absent_log_reads_as_empty_rather_than_failing() {
337 let (_dir, store, a, r) = store();
338 assert!(store.read_log(&a, &r, "deadbeef").unwrap().is_empty());
339 }
340
341 #[test]
342 fn a_run_id_cannot_become_a_path() {
343 let (_dir, store, a, r) = store();
344 for id in ["../../etc/passwd", "a/b", "..", "", "a.json"] {
345 assert!(store.get(&a, &r, id).is_err(), "{id:?} must be refused");
346 }
347 }
348
349 #[test]
350 fn removing_a_repository_removes_its_runs() {
351 let (_dir, store, a, r) = store();
352 let run = Run::queued("alice", "site", "refs/heads/main", "a");
353 store.put(&a, &r, &run).unwrap();
354
355 store.remove_repo(&a, &r);
356 assert!(store.get(&a, &r, &run.id).unwrap().is_none());
357 assert!(store.queued().is_empty());
358 }
359}