forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
19.4 KiB
1//! Self-update: a green CI run of the forge's own repo replaces the running binary and restarts.
2//!
3//! Event-driven. `maybe_deploy` runs from the runner finish path when a job finishes. Nothing
4//! polls. After the swap the process exits with `RESTART_EXIT_CODE`, and the systemd unit
5//! (`Restart=on-failure`) starts the new binary.
6//!
7//! Runners update the same way. The claim route advertises the running binary's sha256 to a runner
8//! of the same target, and `GET /api/v1/ci/runners/binary` serves the bytes.
9
10use std::path::{Path, PathBuf};
11use std::sync::OnceLock;
12use std::sync::atomic::{AtomicBool, Ordering};
13use std::time::Duration;
14
15use forge_core::ci::{CiRun, RunState};
16use forge_core::config::SelfUpdateConfig;
17use futures_util::StreamExt;
18use futures_util::stream::BoxStream;
19use http::header::{CONTENT_LENGTH, CONTENT_TYPE, HeaderValue};
20use http_body_util::StreamBody;
21use sha2::{Digest, Sha256};
22use tokio::io::AsyncWriteExt;
23use tokio_util::io::ReaderStream;
24use topcoat::router::{Body, HeaderMap, StatusCode, response::Response};
25
26use crate::state::AppState;
27
28/// Exit code after a swap. systemd `Restart=on-failure` restarts the unit on any non-zero exit.
29pub const RESTART_EXIT_CODE: u8 = 75;
30/// Response header carrying the sha256 of the running binary, lowercase hex.
31pub const SHA_HEADER: &str = "Forge-Binary-Sha256";
32/// Request header a runner sends with its `<arch>-<os>` target.
33pub const TARGET_HEADER: &str = "Forge-Runner-Target";
34/// How long the staged binary may take to answer `--version`.
35const SANITY_TIMEOUT: Duration = Duration::from_secs(10);
36
37/// Serialises deploys. A deploy that finds another one running is skipped.
38static DEPLOY_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
39/// Wakes the server so it stops after a swap.
40static RESTART: tokio::sync::Notify = tokio::sync::Notify::const_new();
41static RESTART_REQUESTED: AtomicBool = AtomicBool::new(false);
42
43/// The running binary, hashed once at startup.
44pub struct Running {
45 /// The executable path, used to place the sibling `.prev`, `.bad` and staging files.
46 pub path: PathBuf,
47 pub sha256: String,
48}
49
50static RUNNING: OnceLock<Option<Running>> = OnceLock::new();
51
52/// The running binary, hashed on first use. `None` if it cannot be read, which disables
53/// advertising and serving, and deploys.
54pub fn running() -> Option<&'static Running> {
55 RUNNING
56 .get_or_init(|| match load_running() {
57 Ok(running) => {
58 tracing::info!(sha256 = %running.sha256, "self-update: running binary hashed");
59 Some(running)
60 }
61 Err(e) => {
62 tracing::error!(error = %e, "self-update: cannot hash the running binary");
63 None
64 }
65 })
66 .as_ref()
67}
68
69fn load_running() -> std::io::Result<Running> {
70 let path = std::env::current_exe()?;
71 let sha256 = sha256_reader(std::fs::File::open(image_path())?)?;
72 Ok(Running { path, sha256 })
73}
74
75/// The file that holds the running image. On Linux that is `/proc/self/exe`, which still names the
76/// old inode after a rename over the executable. Elsewhere it is the executable path.
77fn image_path() -> PathBuf {
78 let proc_exe = PathBuf::from("/proc/self/exe");
79 if proc_exe.exists() {
80 proc_exe
81 } else {
82 std::env::current_exe().unwrap_or(proc_exe)
83 }
84}
85
86/// Resolves once a swap has asked the server to stop.
87pub async fn restart_requested() {
88 RESTART.notified().await;
89}
90
91/// Whether a swap asked the process to exit with `RESTART_EXIT_CODE`.
92pub fn restart_pending() -> bool {
93 RESTART_REQUESTED.load(Ordering::SeqCst)
94}
95
96/// `<arch>-<os>` for this build, as a runner reports it in `Forge-Runner-Target`.
97pub fn target() -> String {
98 format!("{}-{}", std::env::consts::ARCH, std::env::consts::OS)
99}
100
101/// Whether a runner's requested target is this build's target.
102fn target_matches(requested: Option<&str>) -> bool {
103 requested == Some(target().as_str())
104}
105
106/// Whether a finished run should deploy. Pure, so the rules can be tested. A run deploys only when
107/// it is a successful push to the configured branch of the configured repo, and it is still the
108/// newest run on that branch. Pull request runs (including fork runs) never deploy.
109pub fn should_deploy(
110 cfg: &SelfUpdateConfig,
111 repo_full_name: &str,
112 run: &CiRun,
113 newest_on_ref: Option<&str>,
114) -> bool {
115 cfg.enabled
116 && matches!(run.state, RunState::Success)
117 && run.trigger == "push"
118 && run.pull_number.is_none()
119 && repo_full_name.eq_ignore_ascii_case(&cfg.repo)
120 && run.git_ref == format!("refs/heads/{}", cfg.branch)
121 && newest_on_ref == Some(run.id.as_str())
122}
123
124/// Why a candidate binary must not be installed, or `None` to install it. `bad_list` is the text
125/// of `<exe>.bad`: one hex sha256 per line, `#` starts a comment.
126pub fn skip_reason(candidate_sha: &str, running_sha: &str, bad_list: &str) -> Option<&'static str> {
127 if candidate_sha.eq_ignore_ascii_case(running_sha) {
128 return Some("same as the running binary");
129 }
130 let listed = bad_list
131 .lines()
132 .map(str::trim)
133 .filter(|line| !line.is_empty() && !line.starts_with('#'))
134 .any(|line| line.eq_ignore_ascii_case(candidate_sha));
135 listed.then_some("listed in the bad list")
136}
137
138/// `<name>.<suffix>` next to `exe`, for example `forge.prev`.
139fn sibling(exe: &Path, suffix: &str) -> PathBuf {
140 let name = exe
141 .file_name()
142 .map(|name| name.to_string_lossy().into_owned())
143 .unwrap_or_default();
144 exe.with_file_name(format!("{name}.{suffix}"))
145}
146
147/// Called when a job finishes. Starts a deploy in the background when the run qualifies.
148pub async fn maybe_deploy(state: &AppState, run: &CiRun) {
149 let cfg = &state.config.self_update;
150 if !cfg.enabled || !matches!(run.state, RunState::Success) {
151 return;
152 }
153 let repo = match state.store.repo_by_id(&run.repo_id).await {
154 Ok(Some(repo)) => repo,
155 Ok(None) => return,
156 Err(e) => {
157 tracing::error!(error = %e, "self-update: loading the repo failed");
158 return;
159 }
160 };
161 let full_name = format!("{}/{}", repo.owner_name, repo.name);
162 let newest = match state
163 .store
164 .latest_run_on_ref(&run.repo_id, &run.git_ref)
165 .await
166 {
167 Ok(newest) => newest,
168 Err(e) => {
169 tracing::error!(error = %e, "self-update: loading the newest run failed");
170 return;
171 }
172 };
173 if !should_deploy(cfg, &full_name, run, newest.as_ref().map(|r| r.id.as_str())) {
174 return;
175 }
176 tracing::info!(run_id = %run.id, "self-update: green push run on the deploy branch");
177 let state = state.clone();
178 let run = run.clone();
179 tokio::spawn(async move { deploy(state, run).await });
180}
181
182/// Installs the run's artifact over the running binary and asks the server to restart. Logs and
183/// returns on any failure, leaving the running binary in place.
184pub async fn deploy(state: AppState, run: CiRun) {
185 let Ok(_guard) = DEPLOY_LOCK.try_lock() else {
186 tracing::info!(run_id = %run.id, "self-update: a deploy is already running, skipping");
187 return;
188 };
189 if let Err(e) = deploy_inner(&state, &run).await {
190 tracing::error!(run_id = %run.id, error = %e, "self-update: deploy failed");
191 }
192}
193
194async fn deploy_inner(state: &AppState, run: &CiRun) -> Result<(), String> {
195 let cfg = &state.config.self_update;
196 let running = running().ok_or("the running binary was not hashed at startup")?;
197
198 let jobs = state
199 .store
200 .jobs_for_run(&run.id)
201 .await
202 .map_err(|e| e.to_string())?;
203 let Some(job) = jobs.iter().find(|job| job.name == cfg.job) else {
204 tracing::info!(run_id = %run.id, job = %cfg.job, "self-update: no such job in the run, skipping");
205 return Ok(());
206 };
207 let artifact = state
208 .store
209 .artifact(&job.id, &cfg.artifact)
210 .await
211 .map_err(|e| e.to_string())?;
212 let Some(artifact) = artifact else {
213 tracing::info!(run_id = %run.id, artifact = %cfg.artifact, "self-update: no such artifact, skipping");
214 return Ok(());
215 };
216 let read = state
217 .blobs
218 .read(&crate::ci::artifact_key(&artifact.id))
219 .await
220 .map_err(|e| e.to_string())?
221 .ok_or("the artifact bytes are missing")?;
222
223 // Stage next to the executable, so the final rename stays on one filesystem.
224 let staged = sibling(&running.path, "new");
225 let sha = match write_staged(read.stream, &staged).await {
226 Ok(sha) => sha,
227 Err(e) => {
228 let _ = std::fs::remove_file(&staged);
229 return Err(e);
230 }
231 };
232 if !sha.eq_ignore_ascii_case(&artifact.sha256) {
233 let _ = std::fs::remove_file(&staged);
234 return Err(format!(
235 "artifact sha256 {sha} does not match the recorded {}",
236 artifact.sha256
237 ));
238 }
239 let bad = std::fs::read_to_string(sibling(&running.path, "bad")).unwrap_or_default();
240 if let Some(reason) = skip_reason(&sha, &running.sha256, &bad) {
241 let _ = std::fs::remove_file(&staged);
242 tracing::info!(sha256 = %sha, reason, "self-update: skipped");
243 return Ok(());
244 }
245 if let Err(e) = sanity_check(&staged).await {
246 let _ = std::fs::remove_file(&staged);
247 return Err(e);
248 }
249 if let Err(e) = swap(&running.path, &staged) {
250 let _ = std::fs::remove_file(&staged);
251 return Err(e);
252 }
253 tracing::info!(
254 old_sha256 = %running.sha256,
255 new_sha256 = %sha,
256 exe = %running.path.display(),
257 "self-update: binary replaced, restarting"
258 );
259 RESTART_REQUESTED.store(true, Ordering::SeqCst);
260 RESTART.notify_one();
261 Ok(())
262}
263
264/// Streams the artifact to `path`, hashing as it goes. The file is executable and synced before
265/// this returns. Returns the sha256 in lowercase hex.
266async fn write_staged(
267 mut stream: BoxStream<'static, std::io::Result<bytes::Bytes>>,
268 path: &Path,
269) -> Result<String, String> {
270 use std::os::unix::fs::PermissionsExt;
271
272 let mut file = tokio::fs::File::create(path)
273 .await
274 .map_err(|e| format!("cannot create {}: {e}", path.display()))?;
275 let mut hasher = Sha256::new();
276 while let Some(chunk) = stream.next().await {
277 let chunk = chunk.map_err(|e| e.to_string())?;
278 hasher.update(&chunk);
279 file.write_all(&chunk).await.map_err(|e| e.to_string())?;
280 }
281 file.flush().await.map_err(|e| e.to_string())?;
282 file.sync_all().await.map_err(|e| e.to_string())?;
283 drop(file);
284 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755))
285 .map_err(|e| e.to_string())?;
286 Ok(hex::encode(hasher.finalize()))
287}
288
289/// Runs `<staged> --version` with a timeout. Only exit 0 passes.
290async fn sanity_check(path: &Path) -> Result<(), String> {
291 use std::process::Stdio;
292
293 let mut child = tokio::process::Command::new(path)
294 .arg("--version")
295 .stdin(Stdio::null())
296 .stdout(Stdio::null())
297 .stderr(Stdio::null())
298 .kill_on_drop(true)
299 .spawn()
300 .map_err(|e| format!("the new binary does not run: {e}"))?;
301 match tokio::time::timeout(SANITY_TIMEOUT, child.wait()).await {
302 Ok(Ok(status)) if status.success() => Ok(()),
303 Ok(Ok(status)) => Err(format!("the new binary exited with {status} for --version")),
304 Ok(Err(e)) => Err(format!("waiting for the new binary failed: {e}")),
305 Err(_) => Err("the new binary did not answer --version within 10 seconds".into()),
306 }
307}
308
309/// Keeps the current binary as `<exe>.prev` (a hard link when possible), then renames `staged`
310/// over `exe`.
311fn swap(exe: &Path, staged: &Path) -> Result<(), String> {
312 let prev = sibling(exe, "prev");
313 if prev.exists() {
314 std::fs::remove_file(&prev).map_err(|e| format!("cannot remove the old prev: {e}"))?;
315 }
316 if std::fs::hard_link(exe, &prev).is_err() {
317 std::fs::copy(exe, &prev).map_err(|e| format!("cannot keep the previous binary: {e}"))?;
318 }
319 std::fs::rename(staged, exe).map_err(|e| format!("cannot replace the binary: {e}"))
320}
321
322/// Whether a claim response should advertise the binary: self-update is on and the runner asked
323/// for this target. Adds `Forge-Binary-Sha256` to a 200 or 204 response.
324pub fn advertise(state: &AppState, request: &HeaderMap, res: &mut Response) {
325 if !state.config.self_update.enabled {
326 return;
327 }
328 if !matches!(res.status(), StatusCode::OK | StatusCode::NO_CONTENT) {
329 return;
330 }
331 let requested = request.get(TARGET_HEADER).and_then(|v| v.to_str().ok());
332 if !target_matches(requested) {
333 return;
334 }
335 let Some(running) = running() else {
336 return;
337 };
338 if let Ok(value) = HeaderValue::from_str(&running.sha256) {
339 res.headers_mut().insert(SHA_HEADER, value);
340 }
341}
342
343/// The running binary as a streamed response, with its sha256. The caller checks auth and that
344/// self-update is enabled. `Err` when the binary cannot be read.
345pub async fn binary_response() -> Result<Response, String> {
346 let running = running().ok_or("the running binary was not hashed at startup")?;
347 let file = tokio::fs::File::open(image_path())
348 .await
349 .map_err(|e| e.to_string())?;
350 let len = file.metadata().await.map_err(|e| e.to_string())?.len();
351 let frames = ReaderStream::new(file).map(|chunk| chunk.map(http_body::Frame::data));
352 let mut res = Response::new(Body::new(StreamBody::new(frames)));
353 let headers = res.headers_mut();
354 headers.insert(
355 CONTENT_TYPE,
356 HeaderValue::from_static("application/octet-stream"),
357 );
358 headers.insert(CONTENT_LENGTH, HeaderValue::from(len));
359 if let Ok(value) = HeaderValue::from_str(&running.sha256) {
360 headers.insert(SHA_HEADER, value);
361 }
362 Ok(res)
363}
364
365fn sha256_reader(mut reader: impl std::io::Read) -> std::io::Result<String> {
366 let mut hasher = Sha256::new();
367 let mut buf = vec![0u8; 64 * 1024];
368 loop {
369 let n = reader.read(&mut buf)?;
370 if n == 0 {
371 break;
372 }
373 hasher.update(&buf[..n]);
374 }
375 Ok(hex::encode(hasher.finalize()))
376}
377
378#[cfg(test)]
379mod tests {
380 use super::*;
381 use forge_core::ci::CiRun;
382
383 fn cfg() -> SelfUpdateConfig {
384 SelfUpdateConfig {
385 enabled: true,
386 repo: "Tahinli/forge".into(),
387 branch: "main".into(),
388 job: "build".into(),
389 artifact: "dist_forge".into(),
390 }
391 }
392
393 fn run(trigger: &str, git_ref: &str, state: RunState, pull_number: Option<i64>) -> CiRun {
394 CiRun {
395 id: "run-7".into(),
396 repo_id: "repo-1".into(),
397 number: 7,
398 trigger: trigger.into(),
399 git_ref: git_ref.into(),
400 commit_oid: "a".repeat(40),
401 pull_number,
402 actor_id: "user-1".into(),
403 state,
404 config_error: None,
405 created_at: "now".into(),
406 finished_at: None,
407 }
408 }
409
410 /// A push run of the deploy branch that is still the newest run.
411 fn good() -> CiRun {
412 run("push", "refs/heads/main", RunState::Success, None)
413 }
414
415 #[test]
416 fn push_run_on_the_branch_of_the_repo_deploys() {
417 assert!(should_deploy(
418 &cfg(),
419 "Tahinli/forge",
420 &good(),
421 Some("run-7")
422 ));
423 }
424
425 #[test]
426 fn pull_request_runs_never_deploy() {
427 // A PR run carries a pull number and a refs/pull ref, and its trigger is not "push".
428 let pr = run(
429 "pull_request",
430 "refs/pull/3/head",
431 RunState::Success,
432 Some(3),
433 );
434 assert!(!should_deploy(&cfg(), "Tahinli/forge", &pr, Some("run-7")));
435 // Even if a PR run were stored on the branch ref, the trigger and pull number exclude it.
436 let pr_on_branch = run(
437 "pull_request",
438 "refs/heads/main",
439 RunState::Success,
440 Some(3),
441 );
442 assert!(!should_deploy(
443 &cfg(),
444 "Tahinli/forge",
445 &pr_on_branch,
446 Some("run-7")
447 ));
448 let push_with_pull = run("push", "refs/heads/main", RunState::Success, Some(3));
449 assert!(!should_deploy(
450 &cfg(),
451 "Tahinli/forge",
452 &push_with_pull,
453 Some("run-7")
454 ));
455 }
456
457 #[test]
458 fn other_branch_tag_and_manual_runs_do_not_deploy() {
459 let feature = run("push", "refs/heads/feature", RunState::Success, None);
460 assert!(!should_deploy(
461 &cfg(),
462 "Tahinli/forge",
463 &feature,
464 Some("run-7")
465 ));
466 let tag = run("tag", "refs/tags/v1", RunState::Success, None);
467 assert!(!should_deploy(&cfg(), "Tahinli/forge", &tag, Some("run-7")));
468 let manual = run("manual", "refs/heads/main", RunState::Success, None);
469 assert!(!should_deploy(
470 &cfg(),
471 "Tahinli/forge",
472 &manual,
473 Some("run-7")
474 ));
475 }
476
477 #[test]
478 fn other_repo_does_not_deploy() {
479 assert!(!should_deploy(
480 &cfg(),
481 "Tahinli/other",
482 &good(),
483 Some("run-7")
484 ));
485 // Case differences in the owner or name still match the configured repo.
486 assert!(should_deploy(
487 &cfg(),
488 "tahinli/FORGE",
489 &good(),
490 Some("run-7")
491 ));
492 }
493
494 #[test]
495 fn only_a_successful_run_deploys() {
496 for state in [
497 RunState::Failure,
498 RunState::Cancelled,
499 RunState::Error,
500 RunState::Running,
501 ] {
502 let r = run("push", "refs/heads/main", state, None);
503 assert!(!should_deploy(&cfg(), "Tahinli/forge", &r, Some("run-7")));
504 }
505 }
506
507 #[test]
508 fn an_older_run_on_the_branch_does_not_deploy() {
509 assert!(!should_deploy(
510 &cfg(),
511 "Tahinli/forge",
512 &good(),
513 Some("run-9")
514 ));
515 assert!(!should_deploy(&cfg(), "Tahinli/forge", &good(), None));
516 }
517
518 #[test]
519 fn disabled_config_never_deploys() {
520 let mut off = cfg();
521 off.enabled = false;
522 assert!(!should_deploy(
523 &off,
524 "Tahinli/forge",
525 &good(),
526 Some("run-7")
527 ));
528 }
529
530 #[test]
531 fn same_sha_as_running_is_skipped() {
532 let sha = "ab12";
533 assert_eq!(
534 skip_reason(sha, sha, ""),
535 Some("same as the running binary")
536 );
537 assert_eq!(
538 skip_reason("AB12", sha, ""),
539 Some("same as the running binary")
540 );
541 assert_eq!(skip_reason("cd34", sha, ""), None);
542 }
543
544 #[test]
545 fn bad_list_entries_are_skipped() {
546 let bad = "# rolled back 2026-10-01\n1111\n 2222 \n\n";
547 assert_eq!(
548 skip_reason("2222", "ffff", bad),
549 Some("listed in the bad list")
550 );
551 assert_eq!(
552 skip_reason("1111", "ffff", bad),
553 Some("listed in the bad list")
554 );
555 // A comment line is not a sha.
556 assert_eq!(skip_reason("rolled", "ffff", bad), None);
557 assert_eq!(skip_reason("3333", "ffff", bad), None);
558 }
559
560 #[test]
561 fn target_must_match_this_build() {
562 assert!(target_matches(Some(&target())));
563 assert!(!target_matches(None));
564 assert!(!target_matches(Some("riscv64-plan9")));
565 }
566
567 #[test]
568 fn sha256_of_known_bytes() {
569 let digest = sha256_reader(&b"abc"[..]).unwrap();
570 assert_eq!(
571 digest,
572 "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
573 );
574 }
575}