zuka
zuka/src/ci/mod.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
branchbc985d6bbd321be340e3750b3da4734ee1553cbd ▾
main All branches →
zuka/src/ci/mod.rs
RSmod.rs3.7 KBDownload
1// Continuous integration.
2//
3// A push writes a queued run; an executor in the server process picks it up, checks
4// the revision out into a scratch workspace and runs the steps in the repository's
5// own `.<name>.toml`.
6//
7// The queue is the run records themselves (`ci::run::RunStore::queued`) rather than a
8// separate channel, so a push that lands while the server is down is still picked up
9// when it returns, and the API and the executor cannot disagree about what is
10// pending.
11
12pub mod run;
13pub mod runner;
14pub mod spec;
15
16use crate::git::validate;
17use crate::http::AppState;
18use run::{Run, Status};
19use std::sync::Arc;
20use std::time::Duration;
21
22/// How often the executor looks for queued work.
23///
24/// The hook runs in its own process and writes a record; nothing signals this one.
25/// So this interval is the latency of every run, not just of runs queued while the
26/// process was down. A socket would remove it; the durable record is what makes the
27/// queue survive a restart, and that mattered more.
28const POLL: Duration = Duration::from_secs(5);
29
30/// Drain the queue forever, respecting the per-host concurrency limit.
31pub async fn serve(state: Arc<AppState>) {
32 if !state.config.ci_enabled {
33 return;
34 }
35 eprintln!(
36 "[ci] executor running, {} concurrent",
37 state.config.ci_max_concurrent
38 );
39
40 loop {
41 // Claim what fits, run it, then look again.
42 let capacity = state
43 .config
44 .ci_max_concurrent
45 .saturating_sub(state.running_runs());
46
47 if capacity > 0 {
48 for queued in state.runs.queued().into_iter().take(capacity) {
49 spawn_run(Arc::clone(&state), queued);
50 }
51 }
52 tokio::time::sleep(POLL).await;
53 }
54}
55
56fn spawn_run(state: Arc<AppState>, mut run: Run) {
57 let (Ok(account), Ok(repo)) = (validate::name(&run.account), validate::name(&run.repo)) else {
58 return;
59 };
60
61 // An in-process claim, so two executor ticks in THIS process cannot both pick up
62 // the same run. It is not durable: two processes sharing a data directory would
63 // both run it. Nothing supports that arrangement today, and the control plane
64 // gives each tenant its own.
65 if !state.claim_run(&run.id) {
66 return;
67 }
68
69 tokio::spawn(async move {
70 run.status = Status::Running;
71 run.started_at = Some(crate::account::token::now_secs());
72 let _ = state.runs.put(&account, &repo, &run);
73
74 let outcome = match state.git.repo_path(&account, &repo) {
75 Ok(git_dir) => {
76 runner::execute(&state.config, &state.runs, &account, &repo, &run, &git_dir).await
77 }
78 Err(e) => runner::Outcome {
79 status: Status::Failed,
80 exit_code: None,
81 detail: Some(e.detail_for_user()),
82 },
83 };
84
85 // A cancellation recorded while the run was in flight wins: the operator
86 // asked for it to stop, and reporting "succeeded" afterwards would be a lie.
87 let cancelled = state
88 .runs
89 .get(&account, &repo, &run.id)
90 .ok()
91 .flatten()
92 .is_some_and(|current| current.status == Status::Cancelled);
93
94 run.status = if cancelled {
95 Status::Cancelled
96 } else {
97 outcome.status
98 };
99 run.exit_code = outcome.exit_code;
100 run.detail = outcome.detail;
101 run.finished_at = Some(crate::account::token::now_secs());
102 let _ = state.runs.put(&account, &repo, &run);
103
104 state.release_run(&run.id);
105 eprintln!(
106 "[ci] {}/{} {} -> {:?}",
107 run.account, run.repo, run.id, run.status
108 );
109 });
110}