forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
23.7 KiB
1//! Runner and job API under `/api/v1/ci` (see docs/ci.md, Runner protocol).
2//!
3//! Runners authenticate with `Authorization: Bearer <runner secret>`. A job's routes take
4//! `Authorization: Bearer <job token>`, which `job_by_token` accepts only while the job is Running.
5//! The job token also works as a read-only git credential, see `git_http::job_token_read`.
6//!
7//! Every response is JSON except the empty 204 of an empty claim and the body of a log append.
8
9use forge_ci::config::JobSpec;
10use forge_ci::mask::Masker;
11use forge_ci::proto::{
12 FinishRequest, HeartbeatResponse, JobAssignment, Outcome, RegisterRequest, RegisterResponse,
13};
14use forge_core::Error as CoreError;
15use forge_core::ci::{CiJob, CiRun, CiRunner, JobState};
16use futures_util::StreamExt;
17use http::header::HeaderValue;
18use serde_json::json;
19use std::collections::BTreeMap;
20use topcoat::context::{Cx, app_context};
21use topcoat::router::request::headers;
22use topcoat::router::response::Response;
23use topcoat::router::{Body, HeaderMap, StatusCode, path_param, route};
24
25use crate::api::{self, Refusal};
26use crate::ci::{artifact_key, log_path};
27use crate::pages::repo::https_clone_url;
28use crate::self_update;
29use crate::state::AppState;
30
31path_param!(job_id);
32path_param!(artifact_name);
33
34/// Header carrying the byte offset of a log chunk.
35const LOG_OFFSET_HEADER: &str = "x-forge-log-offset";
36/// Header on a 409 log response carrying the size the server has.
37const LOG_SIZE_HEADER: &str = "x-forge-log-size";
38/// Largest log chunk, in bytes.
39const MAX_LOG_CHUNK: usize = 1024 * 1024;
40/// Largest JSON body on the runner routes, in bytes.
41const MAX_JSON_BODY: usize = 64 * 1024;
42/// How long a claim waits for work before answering 204.
43#[cfg(not(test))]
44const CLAIM_WAIT: std::time::Duration = std::time::Duration::from_secs(25);
45#[cfg(test)]
46const CLAIM_WAIT: std::time::Duration = std::time::Duration::from_millis(50);
47/// Longest runner name or label, in bytes.
48const MAX_NAME_BYTES: usize = 64;
49/// Most labels a runner may register.
50const MAX_LABELS: usize = 32;
51/// Text appended once when the log cap is reached.
52const TRUNCATED_MARKER: &[u8] = b"\n[log truncated]\n";
53const MIB: u64 = 1024 * 1024;
54
55/// Bearer token from the `Authorization` header, or `None`.
56fn bearer_token(headers: &HeaderMap) -> Option<&str> {
57 let value = headers.get(http::header::AUTHORIZATION)?.to_str().ok()?;
58 let (scheme, rest) = value.split_once(' ')?;
59 if !scheme.eq_ignore_ascii_case("bearer") {
60 return None;
61 }
62 let token = rest.trim();
63 (!token.is_empty()).then_some(token)
64}
65
66fn unauthorized() -> Refusal {
67 Box::new(api::error(
68 StatusCode::UNAUTHORIZED,
69 "authentication required",
70 ))
71}
72
73/// The runner behind the Bearer secret. Disabled runners get 403.
74async fn runner_from(state: &AppState, headers: &HeaderMap) -> Result<CiRunner, Refusal> {
75 let Some(secret) = bearer_token(headers) else {
76 return Err(unauthorized());
77 };
78 match state.store.runner_by_secret(secret).await {
79 Ok(Some(runner)) if runner.disabled => Err(Box::new(api::error(
80 StatusCode::FORBIDDEN,
81 "the runner is disabled",
82 ))),
83 Ok(Some(runner)) => Ok(runner),
84 Ok(None) => Err(unauthorized()),
85 Err(e) => Err(api::internal(e)),
86 }
87}
88
89/// The job behind the Bearer job token. Only Running jobs match, so a finished job gets 401.
90async fn job_from(state: &AppState, headers: &HeaderMap, job_id: &str) -> Result<CiJob, Refusal> {
91 let Some(token) = bearer_token(headers) else {
92 return Err(unauthorized());
93 };
94 match state.store.job_by_token(job_id, token).await {
95 Ok(Some(job)) => Ok(job),
96 Ok(None) => Err(unauthorized()),
97 Err(e) => Err(api::internal(e)),
98 }
99}
100
101/// Parses a JSON body of at most [`MAX_JSON_BODY`] bytes.
102async fn json_body<T: serde::de::DeserializeOwned>(body: Body) -> Result<T, Refusal> {
103 let bytes = api::read_body(body, MAX_JSON_BODY).await?;
104 serde_json::from_slice(&bytes).map_err(|e| {
105 Box::new(api::error(
106 StatusCode::BAD_REQUEST,
107 &format!("invalid JSON body: {e}"),
108 ))
109 })
110}
111
112fn valid_name(value: &str) -> bool {
113 !value.trim().is_empty() && value.len() <= MAX_NAME_BYTES && !value.contains(char::is_control)
114}
115
116/// `[A-Za-z0-9._+-]{1,128}`, and not starting with a dot.
117pub fn valid_artifact_name(name: &str) -> bool {
118 !name.is_empty()
119 && name.len() <= 128
120 && !name.starts_with('.')
121 && name
122 .bytes()
123 .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'+' | b'-'))
124}
125
126// ---------------------------------------------------------------------------------------------
127// Registration and claims
128// ---------------------------------------------------------------------------------------------
129
130#[route(POST "/api/v1/ci/runners/register")]
131async fn register(cx: &Cx, body: Body) -> topcoat::Result<Response> {
132 let state = app_context::<AppState>(cx);
133 let request: RegisterRequest = match json_body(body).await {
134 Ok(request) => request,
135 Err(res) => return Ok(*res),
136 };
137 if !valid_name(&request.name) || request.labels.len() > MAX_LABELS {
138 return Ok(api::error(
139 StatusCode::UNPROCESSABLE_ENTITY,
140 "the runner name or labels are invalid",
141 ));
142 }
143 if request.labels.iter().any(|label| !valid_name(label)) {
144 return Ok(api::error(
145 StatusCode::UNPROCESSABLE_ENTITY,
146 "the runner name or labels are invalid",
147 ));
148 }
149 match state
150 .store
151 .register_runner(&request.token, request.name.trim(), &request.labels)
152 .await
153 {
154 Ok((runner, secret)) => {
155 let body = RegisterResponse {
156 runner_id: runner.id,
157 runner_secret: secret,
158 };
159 Ok(api::json(
160 StatusCode::CREATED,
161 &serde_json::to_value(&body).unwrap_or_default(),
162 ))
163 }
164 Err(CoreError::Invalid(_)) => Ok(api::error(
165 StatusCode::UNAUTHORIZED,
166 "the registration token is unknown, used or expired",
167 )),
168 Err(e) => Ok(*api::internal(e)),
169 }
170}
171
172#[route(POST "/api/v1/ci/runners/claim")]
173async fn claim(cx: &Cx) -> topcoat::Result<Response> {
174 let mut res = claim_work(cx).await?;
175 // Advertises the binary sha256 when self-update is on and the runner asked for this target.
176 self_update::advertise(app_context::<AppState>(cx), headers(cx), &mut res);
177 Ok(res)
178}
179
180async fn claim_work(cx: &Cx) -> topcoat::Result<Response> {
181 let state = app_context::<AppState>(cx);
182 // The labels in the request are ignored: the labels stored at registration are used.
183 let runner = match runner_from(state, headers(cx)).await {
184 Ok(runner) => runner,
185 Err(res) => return Ok(*res),
186 };
187
188 // Arm the wake-up before each attempt, so a job queued during the attempt is not missed.
189 // Producers use notify_waiters, which wakes every claimer, so a runner whose labels
190 // don't match a job re-arms and keeps waiting for the rest of CLAIM_WAIT.
191 let waker = state.ci_waker.clone();
192 let deadline = tokio::time::Instant::now() + CLAIM_WAIT;
193 loop {
194 let mut woken = std::pin::pin!(waker.notified());
195 woken.as_mut().enable();
196
197 match try_claim(state, &runner).await {
198 Ok(Some(assignment)) => return Ok(assignment_response(&assignment)),
199 Ok(None) => {}
200 Err(res) => return Ok(*res),
201 }
202 if tokio::time::timeout_at(deadline, woken.as_mut())
203 .await
204 .is_err()
205 {
206 return Ok(empty(StatusCode::NO_CONTENT));
207 }
208 }
209}
210
211/// The running binary, for a runner that updates itself. 404 unless self-update is enabled.
212#[route(GET "/api/v1/ci/runners/binary")]
213async fn runner_binary(cx: &Cx) -> topcoat::Result<Response> {
214 let state = app_context::<AppState>(cx);
215 if let Err(res) = runner_from(state, headers(cx)).await {
216 return Ok(*res);
217 }
218 if !state.config.self_update.enabled {
219 return Ok(api::error(StatusCode::NOT_FOUND, "self-update is disabled"));
220 }
221 match self_update::binary_response().await {
222 Ok(res) => Ok(res),
223 Err(e) => {
224 tracing::error!(error = %e, "ci: serving the runner binary failed");
225 Ok(api::error(
226 StatusCode::INTERNAL_SERVER_ERROR,
227 "the binary is unavailable",
228 ))
229 }
230 }
231}
232
233fn assignment_response(assignment: &JobAssignment) -> Response {
234 api::json(
235 StatusCode::OK,
236 &serde_json::to_value(assignment).unwrap_or_default(),
237 )
238}
239
240fn empty(status: StatusCode) -> Response {
241 let mut res = Response::new(topcoat::router::Body::empty());
242 *res.status_mut() = status;
243 res
244}
245
246/// One claim attempt. Builds the assignment, including the secrets the job may see.
247async fn try_claim(state: &AppState, runner: &CiRunner) -> Result<Option<JobAssignment>, Refusal> {
248 let Some((job, run, job_token)) = state.store.claim_job(runner).await.map_err(api::internal)?
249 else {
250 return Ok(None);
251 };
252 state.ci_live.bump_job(&job.id);
253 state.ci_live.bump_run(&run.id);
254 let spec: JobSpec = serde_json::from_str(&job.spec_json).map_err(api::internal)?;
255 let repo = state
256 .store
257 .repo_by_id(&run.repo_id)
258 .await
259 .map_err(api::internal)?
260 .ok_or_else(|| api::internal("claimed job belongs to a missing repo"))?;
261
262 let secrets = if is_fork_run(state, &run).await? {
263 BTreeMap::new()
264 } else {
265 state
266 .store
267 .ci_secret_values(&run.repo_id, &spec.secrets)
268 .await
269 .map_err(api::internal)?
270 };
271
272 Ok(Some(JobAssignment {
273 job_id: job.id,
274 job_token,
275 repo: format!("{}/{}", repo.owner_name, repo.name),
276 clone_url: https_clone_url(&state.config, &repo.owner_name, &repo.name),
277 commit: run.commit_oid,
278 git_ref: run.git_ref,
279 job_name: job.name,
280 image: spec.image,
281 run: spec.run,
282 env: spec.env,
283 secrets,
284 artifacts: spec.artifacts,
285 timeout_minutes: spec.timeout_minutes,
286 network: spec.network,
287 }))
288}
289
290/// True for a pull-request run whose head is not this repo, or whose pull request is gone. Such
291/// runs get no secrets. Missing data counts as a fork, so the answer errs on the safe side.
292async fn is_fork_run(state: &AppState, run: &CiRun) -> Result<bool, Refusal> {
293 if run.trigger != "pull_request" {
294 return Ok(false);
295 }
296 let Some(number) = run.pull_number else {
297 return Ok(true);
298 };
299 match state
300 .store
301 .pull_by_number(&run.repo_id, number)
302 .await
303 .map_err(api::internal)?
304 {
305 Some(pull) => Ok(pull.head_repo_id.as_deref() != Some(run.repo_id.as_str())),
306 None => Ok(true),
307 }
308}
309
310// ---------------------------------------------------------------------------------------------
311// Heartbeat, logs and artifacts
312// ---------------------------------------------------------------------------------------------
313
314#[route(POST "/api/v1/ci/jobs/{job_id}/heartbeat")]
315async fn heartbeat(cx: &Cx) -> topcoat::Result<Response> {
316 let state = app_context::<AppState>(cx);
317 let job = match job_from(state, headers(cx), path_param::<JobId>(cx)).await {
318 Ok(job) => job,
319 Err(res) => return Ok(*res),
320 };
321 match state.store.heartbeat(&job.id).await {
322 Ok(cancel) => Ok(api::json(
323 StatusCode::OK,
324 &serde_json::to_value(HeartbeatResponse { cancel }).unwrap_or_default(),
325 )),
326 Err(e) => Ok(*api::internal(e)),
327 }
328}
329
330/// Outcome of a log append.
331enum LogWrite {
332 /// The chunk is stored (or was dropped by the cap); the size is the new byte count.
333 Stored(u64),
334 /// This exact chunk was already stored: a retry. The size is unchanged.
335 Duplicate(u64),
336 /// The offset does not match the stored size.
337 Mismatch(u64),
338}
339
340#[route(POST "/api/v1/ci/jobs/{job_id}/logs")]
341async fn append_log(cx: &Cx, body: Body) -> topcoat::Result<Response> {
342 let state = app_context::<AppState>(cx);
343 let headers = headers(cx);
344 let job = match job_from(state, headers, path_param::<JobId>(cx)).await {
345 Ok(job) => job,
346 Err(res) => return Ok(*res),
347 };
348 let Some(offset) = headers
349 .get(LOG_OFFSET_HEADER)
350 .and_then(|value| value.to_str().ok())
351 .and_then(|value| value.parse::<u64>().ok())
352 else {
353 return Ok(api::error(
354 StatusCode::BAD_REQUEST,
355 "X-Forge-Log-Offset must be a byte offset",
356 ));
357 };
358 let chunk = match api::read_body(body, MAX_LOG_CHUNK).await {
359 Ok(chunk) => chunk,
360 Err(res) => return Ok(*res),
361 };
362 match write_log(state, &job, offset, &chunk).await {
363 Ok(LogWrite::Stored(size) | LogWrite::Duplicate(size)) => {
364 Ok(api::json(StatusCode::OK, &json!({ "size": size })))
365 }
366 Ok(LogWrite::Mismatch(size)) => {
367 let mut res = api::error(StatusCode::CONFLICT, "the log offset does not match");
368 res.headers_mut().insert(
369 http::HeaderName::from_static(LOG_SIZE_HEADER),
370 HeaderValue::from(size),
371 );
372 Ok(res)
373 }
374 Err(res) => Ok(*res),
375 }
376}
377
378/// Appends one chunk of raw log bytes. Offsets count the raw bytes the runner sent, which is
379/// `job.log_bytes`; the file holds the masked text, so its length may differ.
380async fn write_log(
381 state: &AppState,
382 job: &CiJob,
383 offset: u64,
384 chunk: &[u8],
385) -> Result<LogWrite, Refusal> {
386 let size = u64::try_from(job.log_bytes).unwrap_or(0);
387 let len = chunk.len() as u64;
388 if offset != size {
389 return Ok(if offset.checked_add(len) == Some(size) {
390 LogWrite::Duplicate(size)
391 } else {
392 LogWrite::Mismatch(size)
393 });
394 }
395 if chunk.is_empty() {
396 return Ok(LogWrite::Stored(size));
397 }
398
399 let masked = mask_for_job(state, job, chunk).await?;
400 let path = log_path(&state.config.data_dir, &job.id);
401 store_log_bytes(state, &path, &masked).await?;
402 state
403 .store
404 .add_log_bytes(&job.id, len as i64)
405 .await
406 .map_err(api::internal)?;
407 state.ci_live.bump_job(&job.id);
408 Ok(LogWrite::Stored(size + len))
409}
410
411/// Masks a chunk with the values of the job's secrets. The chunk is masked on its own, so a
412/// secret split across two requests is masked by the runner, which holds back the boundary.
413async fn mask_for_job(state: &AppState, job: &CiJob, chunk: &[u8]) -> Result<Vec<u8>, Refusal> {
414 let spec: JobSpec = serde_json::from_str(&job.spec_json).map_err(api::internal)?;
415 if spec.secrets.is_empty() {
416 return Ok(chunk.to_vec());
417 }
418 let Some(run) = state
419 .store
420 .run_by_id(&job.run_id)
421 .await
422 .map_err(api::internal)?
423 else {
424 return Err(api::internal("log for a job with no run"));
425 };
426 let values = state
427 .store
428 .ci_secret_values(&run.repo_id, &spec.secrets)
429 .await
430 .map_err(api::internal)?;
431 let masker = Masker::new(values.into_values());
432 let mut stream = masker.stream();
433 let mut out = stream.push(chunk);
434 out.extend(stream.finish());
435 Ok(out)
436}
437
438/// The bytes of `masked` that may be appended to a log of `current` bytes under a `cap`. Once the
439/// log reaches the cap nothing more is kept. When the new bytes do not fit, the part that does is
440/// kept and the marker follows it, so the file ends at exactly `cap` bytes.
441pub fn fit_to_cap(current: u64, cap: u64, masked: &[u8]) -> Vec<u8> {
442 if current >= cap {
443 return Vec::new();
444 }
445 let data_cap = cap.saturating_sub(TRUNCATED_MARKER.len() as u64);
446 let room = data_cap.saturating_sub(current);
447 if masked.len() as u64 <= room {
448 return masked.to_vec();
449 }
450 let mut out = masked[..room as usize].to_vec();
451 out.extend_from_slice(TRUNCATED_MARKER);
452 out
453}
454
455/// Appends masked bytes to the log file, stopping at the size cap with one marker line.
456async fn store_log_bytes(
457 state: &AppState,
458 path: &std::path::Path,
459 masked: &[u8],
460) -> Result<(), Refusal> {
461 use tokio::io::AsyncWriteExt;
462
463 let cap = state.config.ci.max_log_mb.saturating_mul(MIB);
464 let current = match tokio::fs::metadata(path).await {
465 Ok(meta) => meta.len(),
466 Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0,
467 Err(e) => return Err(api::internal(e)),
468 };
469 let out = fit_to_cap(current, cap, masked);
470 if out.is_empty() {
471 return Ok(());
472 }
473
474 if let Some(parent) = path.parent() {
475 tokio::fs::create_dir_all(parent)
476 .await
477 .map_err(api::internal)?;
478 }
479 let mut file = tokio::fs::OpenOptions::new()
480 .create(true)
481 .append(true)
482 .open(path)
483 .await
484 .map_err(api::internal)?;
485 file.write_all(&out).await.map_err(api::internal)?;
486 file.flush().await.map_err(api::internal)?;
487 Ok(())
488}
489
490#[route(PUT "/api/v1/ci/jobs/{job_id}/artifacts/{artifact_name}")]
491async fn upload_artifact(cx: &Cx, body: Body) -> topcoat::Result<Response> {
492 let state = app_context::<AppState>(cx);
493 let job = match job_from(state, headers(cx), path_param::<JobId>(cx)).await {
494 Ok(job) => job,
495 Err(res) => return Ok(*res),
496 };
497 let name = path_param::<ArtifactName>(cx);
498 if !valid_artifact_name(name) {
499 return Ok(api::error(
500 StatusCode::UNPROCESSABLE_ENTITY,
501 "the artifact name must be 1 to 128 of A-Z a-z 0-9 . _ + - and not start with a dot",
502 ));
503 }
504 match upload(state, &job, name, body).await {
505 Ok(artifact_id) => Ok(api::json(
506 StatusCode::CREATED,
507 &json!({ "id": artifact_id, "name": name }),
508 )),
509 Err(res) => Ok(*res),
510 }
511}
512
513/// Streams the body to a temporary blob, then records the artifact and moves the blob to its
514/// final key `ci/<artifact id>`. The id is only known once the row exists.
515async fn upload(state: &AppState, job: &CiJob, name: &str, body: Body) -> Result<String, Refusal> {
516 let max_artifacts = state.config.ci.max_artifacts_per_job;
517 let existing = state
518 .store
519 .list_artifacts(&job.id)
520 .await
521 .map_err(api::internal)?;
522 if existing.len() as u64 >= u64::from(max_artifacts) {
523 return Err(Box::new(api::error(
524 StatusCode::CONFLICT,
525 "the job has too many artifacts",
526 )));
527 }
528 if existing.iter().any(|artifact| artifact.name == name) {
529 return Err(Box::new(api::error(
530 StatusCode::CONFLICT,
531 "an artifact with this name already exists",
532 )));
533 }
534
535 let max_bytes = state.config.ci.max_artifact_mb.saturating_mul(MIB);
536 let temp_key = format!("ci/tmp/{}", ulid::Ulid::generate());
537 let mut writer = state
538 .blobs
539 .begin(&temp_key, max_bytes)
540 .await
541 .map_err(api::internal)?;
542 let mut stream = std::pin::pin!(body.into_data_stream());
543 while let Some(next) = stream.next().await {
544 let outcome = match next {
545 Ok(chunk) => writer.push(&chunk).await.map_err(|e| (e, false)),
546 Err(_) => Err((CoreError::Invalid("upload interrupted".into()), true)),
547 };
548 if let Err((e, interrupted)) = outcome {
549 writer.abort().await;
550 return Err(upload_failure(e, interrupted));
551 }
552 }
553 let (size, sha256) = match writer.finish().await {
554 Ok(done) => done,
555 Err(e) => {
556 discard_temp(state, &temp_key).await;
557 return Err(api::internal(e));
558 }
559 };
560
561 let artifact = match state
562 .store
563 .add_artifact(&job.id, name, size as i64, &sha256)
564 .await
565 {
566 Ok(artifact) => artifact,
567 Err(e) => {
568 discard_temp(state, &temp_key).await;
569 return Err(match e {
570 CoreError::Invalid(message) => Box::new(api::error(StatusCode::CONFLICT, &message)),
571 other => api::internal(other),
572 });
573 }
574 };
575 let final_key = artifact_key(&artifact.id);
576 if let Err(e) = state.blobs.rename(&temp_key, &final_key).await {
577 discard_temp(state, &temp_key).await;
578 return Err(api::internal(e));
579 }
580 Ok(artifact.id)
581}
582
583/// Removes a temporary blob. A failure is logged; the blob is unreachable either way.
584async fn discard_temp(state: &AppState, key: &str) {
585 if let Err(e) = state.blobs.delete(key).await {
586 tracing::error!(error = %e, "ci: could not remove a temporary artifact blob");
587 }
588}
589
590fn upload_failure(e: CoreError, interrupted: bool) -> Refusal {
591 match e {
592 CoreError::Invalid(_) if !interrupted => Box::new(api::error(
593 StatusCode::PAYLOAD_TOO_LARGE,
594 "the artifact is larger than the upload limit",
595 )),
596 CoreError::Invalid(message) => Box::new(api::error(StatusCode::BAD_REQUEST, &message)),
597 other => api::internal(other),
598 }
599}
600
601// ---------------------------------------------------------------------------------------------
602// Finish
603// ---------------------------------------------------------------------------------------------
604
605#[route(POST "/api/v1/ci/jobs/{job_id}/finish")]
606async fn finish(cx: &Cx, body: Body) -> topcoat::Result<Response> {
607 let state = app_context::<AppState>(cx);
608 let job = match job_from(state, headers(cx), path_param::<JobId>(cx)).await {
609 Ok(job) => job,
610 Err(res) => return Ok(*res),
611 };
612 let request: FinishRequest = match json_body(body).await {
613 Ok(request) => request,
614 Err(res) => return Ok(*res),
615 };
616 let state_after = match request.outcome {
617 Outcome::Success => JobState::Success,
618 Outcome::Failure => JobState::Failure,
619 Outcome::Cancelled => JobState::Cancelled,
620 Outcome::Error => JobState::Error,
621 };
622 match finish_job(state, &job, state_after, &request).await {
623 Ok(state_name) => Ok(api::json(StatusCode::OK, &json!({ "state": state_name }))),
624 Err(res) => Ok(*res),
625 }
626}
627
628/// Finishes the job, then syncs the statuses of the job and of every job whose state changed
629/// (skipped or queued by this finish). Wakes the claimers when something was queued, and the
630/// watchers of every job and run that changed.
631async fn finish_job(
632 state: &AppState,
633 job: &CiJob,
634 outcome: JobState,
635 request: &FinishRequest,
636) -> Result<&'static str, Refusal> {
637 let before: BTreeMap<String, JobState> = state
638 .store
639 .jobs_for_run(&job.run_id)
640 .await
641 .map_err(api::internal)?
642 .into_iter()
643 .map(|other| (other.id, other.state))
644 .collect();
645 let exit_code = request.exit_code.map(i64::from);
646 let run = state
647 .store
648 .finish_job(&job.id, outcome, exit_code, request.message.as_deref())
649 .await
650 .map_err(api::internal)?;
651
652 let mut queued = false;
653 let mut any_changed = false;
654 for other in state
655 .store
656 .jobs_for_run(&job.run_id)
657 .await
658 .map_err(api::internal)?
659 {
660 let changed = before.get(&other.id) != Some(&other.state);
661 if changed {
662 any_changed = true;
663 if other.state.is_terminal() {
664 state.ci_live.close_job(&other.id);
665 } else {
666 state.ci_live.bump_job(&other.id);
667 }
668 }
669 if other.id == job.id
670 || (changed && matches!(other.state, JobState::Skipped | JobState::Queued))
671 {
672 queued |= other.state == JobState::Queued;
673 crate::ci::sync_status(state, &run, &other).await;
674 }
675 }
676 if any_changed {
677 state.ci_live.bump_run(&run.id);
678 }
679 if queued {
680 state.ci_waker.notify_waiters();
681 }
682 // A run that just went green may deploy the forge itself.
683 self_update::maybe_deploy(state, &run).await;
684 Ok(run.state.as_str())
685}
686
687#[cfg(test)]
688#[path = "api_tests.rs"]
689mod tests;