forge
ExploreLog in

Tahinli/forge

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