0 0 🚀 feat: self-update on green ci and repo workflow Tahinli/forge
🚀 feat: self-update on green ci and repo workflow
- forge swaps its binary when a push to main passes CI (artifact dist_forge), keeps .prev, exits 75 for systemd restart
- runners update from the server's binary when idle via claim headers
- systemd user units with OnFailure rollback and a .bad list
- .forge/ci.toml runs test and build
44 changed files +1597 -74 @@ -0,0 +1,32 @@ 1 +# CI for forge itself. Format: docs/ci.md, parser: crates/forge-ci/src/config.rs. 2 +# 3 +# cmake is needed because aws-lc-sys (via aws-lc-rs, pulled in by rustls) builds its C 4 +# sources with cmake. The rust image has a C compiler but not cmake. 5 +# The workflow has no branch condition, so the build job also runs for pull requests. 6 + 7 +[on] 8 +push = ["main"] 9 +pull_request = true 10 +manual = true 11 + 12 +[jobs.test] 13 +image = "docker.io/library/rust:1" 14 +run = """ 15 +apt-get update 16 +apt-get install -y --no-install-recommends cmake 17 +cargo test --workspace 18 +""" 19 +timeout_minutes = 60 20 + 21 +[jobs.build] 22 +image = "docker.io/library/rust:1" 23 +needs = ["test"] 24 +run = """ 25 +apt-get update 26 +apt-get install -y --no-install-recommends cmake 27 +cargo build --profile dist -p forge-web 28 +mkdir -p dist 29 +cp target/dist/forge dist/forge 30 +""" 31 +artifacts = ["dist/forge"] 32 +timeout_minutes = 60
@@ -114,6 +114,10 @@ similar = "3.2.0" 114 114 semver = "1.0.28" 115 115 zip = { version = "9.0.3", default-features = false, features = ["deflate"] } 116 116 117 +[profile.dev.package.sha2] 118 +# self-update hashes the ~1 GB debug binary at boot; unoptimized sha2 takes ~26 s. 119 +opt-level = 3 120 + 117 121 [profile.release] 118 122 lto = true 119 123 codegen-units = 1
@@ -110,3 +110,18 @@ trusted_proxies = [] 110 110 # [orgs] 111 111 # Who may create organizations: "all" (any signed-in person) or "admin" (instance admins only). 112 112 # creation = "all" 113 + 114 +# Self-update (off by default). When enabled, a successful push run of `repo` on `branch` replaces 115 +# the running binary with the artifact of `job` and restarts. Run under systemd with Restart=on-failure, 116 +# because the process exits with code 75 after the swap. A sha256 listed in `<exe>.bad` is never installed. 117 +# [self_update] 118 +# Whether a green run deploys its artifact. 119 +# enabled = false 120 +# The repo whose runs deploy, as owner/name. 121 +# repo = "Tahinli/forge" 122 +# The branch whose push runs deploy. 123 +# branch = "main" 124 +# The CI job that produces the artifact. 125 +# job = "build" 126 +# The artifact name produced by job. A single file. CI flattens paths, so dist/forge is stored as dist_forge. 127 +# artifact = "dist_forge"
@@ -1,6 +1,6 @@ 1 1 -- CI: runners, registration tokens, runs, jobs, artifacts and sealed repo secrets. 2 2 -- Runner, registration and job tokens are stored as sha256 hex only. Timestamps are RFC 3339 3 --- UTC stamps from time.rs. See docs/ci-interfaces.md section 2. 3 +-- UTC stamps from time.rs. 4 4 5 5 CREATE TABLE ci_runners ( 6 6 id TEXT PRIMARY KEY,
@@ -1,5 +1,5 @@ 1 1 -- Security advisories (OSV dumps), the dependencies read from each repo's lockfiles, and the 2 --- alerts that match them. See docs/wave-c-interfaces.md (A1). 2 +-- alerts that match them. 3 3 CREATE TABLE advisories ( 4 4 id TEXT NOT NULL, 5 5 ecosystem TEXT NOT NULL,
@@ -1,5 +1,4 @@ 1 --- One inbox row per admin and repo, for security alerts a lockfile scan opened (see 2 --- docs/wave-c-interfaces.md, F-A). Alerts are not issues, so they cannot use the `notifications` 1 +-- One inbox row per admin and repo, for security alerts a lockfile scan opened. Alerts are not issues, so they cannot use the `notifications` 3 2 -- table, which points at issues. `new_alerts` counts the alerts opened since the row was last read. 4 3 CREATE TABLE advisory_notifications ( 5 4 user_id TEXT NOT NULL REFERENCES users(id),
@@ -99,6 +99,65 @@ pub struct Config { 99 99 /// Organization rules, from the `[orgs]` section. 100 100 #[serde(default)] 101 101 pub orgs: OrgsConfig, 102 + /// Self-update after a green CI run of the forge's own repo, from the `[self_update]` section. 103 + #[serde(default)] 104 + pub self_update: SelfUpdateConfig, 105 +} 106 + 107 +/// Self-update, from the `[self_update]` section. When enabled, a successful push run of `repo` 108 +/// on `branch` replaces the running binary with the `artifact` of `job` and restarts. 109 +#[derive(Debug, Clone, Deserialize)] 110 +#[serde(deny_unknown_fields, default)] 111 +pub struct SelfUpdateConfig { 112 + /// Whether a green run deploys its artifact. Off by default. 113 + pub enabled: bool, 114 + /// The repo whose runs deploy, as `owner/name`. 115 + pub repo: String, 116 + /// The branch whose push runs deploy. 117 + pub branch: String, 118 + /// The CI job that produces the artifact. 119 + pub job: String, 120 + /// The artifact name produced by `job`. A single file. CI flattens paths, so `dist/forge` 121 + /// is stored as `dist_forge`. 122 + pub artifact: String, 123 +} 124 + 125 +impl Default for SelfUpdateConfig { 126 + fn default() -> Self { 127 + SelfUpdateConfig { 128 + enabled: false, 129 + repo: String::new(), 130 + branch: "main".into(), 131 + job: "build".into(), 132 + artifact: "dist_forge".into(), 133 + } 134 + } 135 +} 136 + 137 +impl SelfUpdateConfig { 138 + fn validate(&self) -> Result<()> { 139 + if !self.enabled { 140 + return Ok(()); 141 + } 142 + let parts: Vec<&str> = self.repo.split('/').collect(); 143 + if parts.len() != 2 || parts.iter().any(|part| part.is_empty()) { 144 + return Err(Error::Config( 145 + "[self_update] repo must be owner/name when enabled".into(), 146 + )); 147 + } 148 + for (field, value) in [ 149 + ("branch", &self.branch), 150 + ("job", &self.job), 151 + ("artifact", &self.artifact), 152 + ] { 153 + if value.trim().is_empty() { 154 + return Err(Error::Config(format!( 155 + "[self_update] {field} must not be empty when enabled" 156 + ))); 157 + } 158 + } 159 + Ok(()) 160 + } 102 161 } 103 162 104 163 /// Organization rules, from the `[orgs]` section. @@ -299,7 +358,7 @@ impl Default for ReleasesConfig { 299 358 } 300 359 } 301 360 302 -/// Scheduled backups, from the `[backup]` section. See `docs/backup-interfaces.md`. 361 +/// Scheduled backups, from the `[backup]` section. See `docs/backup.md`. 303 362 #[derive(Debug, Clone, Deserialize)] 304 363 #[serde(deny_unknown_fields, default)] 305 364 pub struct BackupConfig { @@ -395,6 +454,7 @@ impl Config { 395 454 toml::from_str(&text).map_err(|e| Error::Config(format!("{}: {e}", path.display())))?; 396 455 config.backup.validate()?; 397 456 config.replication.validate()?; 457 + config.self_update.validate()?; 398 458 Ok(config) 399 459 } 400 460 } @@ -548,6 +608,41 @@ mod tests { 548 608 assert!(matches!(Config::load(&path), Err(Error::Config(_)))); 549 609 } 550 610 611 + #[test] 612 + fn self_update_section_defaults_to_disabled() { 613 + let dir = tempfile::tempdir().unwrap(); 614 + let path = dir.path().join("forge.toml"); 615 + std::fs::write(&path, DEV_DEFAULTS).unwrap(); 616 + let cfg = Config::load(&path).unwrap(); 617 + assert!(!cfg.self_update.enabled); 618 + assert_eq!(cfg.self_update.branch, "main"); 619 + assert_eq!(cfg.self_update.job, "build"); 620 + assert_eq!(cfg.self_update.artifact, "dist_forge"); 621 + } 622 + 623 + #[test] 624 + fn self_update_enabled_requires_owner_name_repo() { 625 + let dir = tempfile::tempdir().unwrap(); 626 + let path = dir.path().join("forge.toml"); 627 + let base = format!("{DEV_DEFAULTS}\n[self_update]\nenabled = true\n"); 628 + for repo in ["", "forge", "a/b/c", "/forge", "owner/"] { 629 + std::fs::write(&path, format!("{base}repo = \"{repo}\"\n")).unwrap(); 630 + assert!(Config::load(&path).is_err(), "repo `{repo}` should be refused"); 631 + } 632 + std::fs::write(&path, format!("{base}repo = \"Tahinli/forge\"\n")).unwrap(); 633 + let cfg = Config::load(&path).unwrap(); 634 + assert!(cfg.self_update.enabled); 635 + assert_eq!(cfg.self_update.repo, "Tahinli/forge"); 636 + } 637 + 638 + #[test] 639 + fn self_update_rejects_unknown_keys() { 640 + let dir = tempfile::tempdir().unwrap(); 641 + let path = dir.path().join("forge.toml"); 642 + std::fs::write(&path, format!("{DEV_DEFAULTS}\n[self_update]\nbogus = 1\n")).unwrap(); 643 + assert!(Config::load(&path).is_err()); 644 + } 645 + 551 646 #[test] 552 647 fn backup_section_is_parsed_and_validated() { 553 648 let dir = tempfile::tempdir().unwrap();
@@ -1,4 +1,4 @@ 1 -//! Short-lived signed tokens for Git LFS transfer actions (see docs/wave-b-interfaces.md). 1 +//! Short-lived signed tokens for Git LFS transfer actions. 2 2 //! 3 3 //! Token format: `"<expires>.<op>.<base64url-nopad(hmac_sha256)>"`, where the MAC covers 4 4 //! `"lfs\0" + repo_id + "\0" + op + "\0" + expires`. `op` is `download`, `upload` or `any`.
@@ -1,4 +1,4 @@ 1 -//! Domain types. Stored shapes are described in docs/m1-interfaces.md. 1 +//! Domain types. Stored shapes are described in docs/architecture.md, section 4. 2 2 3 3 pub type Id = String; 4 4
@@ -1,4 +1,4 @@ 1 -//! Signed tokens for private raw-file URLs (see docs/m2-interfaces.md). 1 +//! Signed tokens for private raw-file URLs. 2 2 //! 3 3 //! Token format: `"<expires>.<base64url-nopad(hmac_sha256)>"`, where the MAC covers 4 4 //! `"raw-v1:" + repo_id + ":" + expires`.
@@ -1,4 +1,4 @@ 1 -//! In-process repo reads via gix (see docs/m2-interfaces.md). 1 +//! In-process repo reads via gix. 2 2 //! 3 3 //! Everything here is blocking; callers run it inside `spawn_blocking`. Repositories are opened 4 4 //! isolated: the environment and user/system git configuration are ignored.
@@ -1,4 +1,4 @@ 1 -//! git CLI history: log, commit diff, last commits, archive (see docs/m2-interfaces.md). 1 +//! git CLI history: log, commit diff, last commits, archive. 2 2 //! Every commit argument is a validated full hex OID; paths go after `--` as literal pathspecs. 3 3 4 4 use std::collections::{HashMap, HashSet};
@@ -18,8 +18,9 @@ use crate::client::ForgeClient; 18 18 use crate::config::{self, ExecutorKind, ImagesConfig, Limits, RunnerConfig, UsernsMode}; 19 19 use crate::executor::{BuiltinExecutor, Executor, PodmanExecutor}; 20 20 use crate::oci::ImageStore; 21 -use crate::runner::{RunOptions, Runner}; 21 +use crate::runner::{RunEnd, RunOptions, Runner}; 22 22 use crate::sandbox::idmap::IdMap; 23 +use crate::update; 23 24 24 25 /// How often unused images are garbage collected, after the one at startup. 25 26 const GC_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); @@ -124,7 +125,7 @@ where 124 125 }; 125 126 runtime.block_on(async { 126 127 match run(cli).await { 127 - Ok(()) => ExitCode::SUCCESS, 128 + Ok(code) => code, 128 129 Err(e) => { 129 130 eprintln!("{bin_name}: {e}"); 130 131 ExitCode::FAILURE @@ -166,7 +167,7 @@ fn hidden_dispatch(args: &[OsString]) -> i32 { 166 167 } 167 168 } 168 169 169 -async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> { 170 +async fn run(cli: Cli) -> Result<ExitCode, Box<dyn std::error::Error>> { 170 171 match cli.command { 171 172 Command::Register { 172 173 url, @@ -205,6 +206,7 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> { 205 206 images: ImagesConfig::default(), 206 207 sandbox: config::SandboxConfig::default(), 207 208 registries: Vec::new(), 209 + auto_update: true, 208 210 }; 209 211 config::save(&path, &cfg)?; 210 212 println!( @@ -212,7 +214,7 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> { 212 214 resp.runner_id, 213 215 path.display() 214 216 ); 215 - Ok(()) 217 + Ok(ExitCode::SUCCESS) 216 218 } 217 219 Command::Run { config: path } => { 218 220 let path = match path { @@ -255,8 +257,11 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> { 255 257 stop.cancel(); 256 258 } 257 259 }); 258 - runner.run(shutdown).await; 259 - Ok(()) 260 + match runner.run(shutdown).await { 261 + RunEnd::Stopped => Ok(ExitCode::SUCCESS), 262 + // systemd restarts the runner on this code, and it starts the new binary. 263 + RunEnd::Updated => Ok(ExitCode::from(update::RESTART_EXIT_CODE)), 264 + } 260 265 } 261 266 Command::Check { 262 267 work_dir, @@ -269,7 +274,7 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> { 269 274 let mode = check_userns_mode(config_path)?; 270 275 let helper = std::env::current_exe()?; 271 276 if crate::check::run(&helper, &work_dir, mode).await { 272 - Ok(()) 277 + Ok(ExitCode::SUCCESS) 273 278 } else { 274 279 Err("some host checks failed, see the report above".into()) 275 280 }
@@ -11,16 +11,25 @@ use forge_ci::proto::{ 11 11 RegisterResponse, 12 12 }; 13 13 use reqwest::StatusCode; 14 -use reqwest::header::CONTENT_LENGTH; 14 +use reqwest::header::{CONTENT_LENGTH, HeaderMap}; 15 15 use serde::de::DeserializeOwned; 16 +use sha2::{Digest, Sha256}; 17 +use tokio::io::AsyncWriteExt; 16 18 17 19 /// The claim long-poll is 25s on the forge side, so 35s leaves room for the response. 18 20 pub const CLAIM_TIMEOUT: Duration = Duration::from_secs(35); 19 21 const REQUEST_TIMEOUT: Duration = Duration::from_secs(60); 20 22 const LOG_TIMEOUT: Duration = Duration::from_secs(60); 21 23 const ARTIFACT_TIMEOUT: Duration = Duration::from_secs(60 * 60); 24 +const BINARY_TIMEOUT: Duration = Duration::from_secs(10 * 60); 22 25 const HEADER_LOG_OFFSET: &str = "X-Forge-Log-Offset"; 23 26 const HEADER_LOG_SIZE: &str = "X-Forge-Log-Size"; 27 +/// Sent on every claim: `<ARCH>-<OS>` of this runner, for example `x86_64-linux`. 28 +const HEADER_RUNNER_TARGET: &str = "Forge-Runner-Target"; 29 +/// On a claim response: the sha256 (hex) of the forge's own binary, if the forge advertises one. 30 +const HEADER_BINARY_SHA256: &str = "Forge-Binary-Sha256"; 31 +/// `GET`: the forge's binary, for a runner that is updating itself. 32 +const BINARY_PATH: &str = "/api/v1/ci/runners/binary"; 24 33 25 34 /// Errors from talking to the forge. 26 35 #[derive(Debug, thiserror::Error)] @@ -44,6 +53,15 @@ pub enum LogAck { 44 53 Resync(u64), 45 54 } 46 55 56 +/// The forge's answer to a claim. 57 +#[derive(Clone, Debug)] 58 +pub struct ClaimReply { 59 + /// The job assigned to this runner, or `None` when there is nothing to run (204). 60 + pub job: Option<JobAssignment>, 61 + /// The sha256 (lowercase hex) of the forge's binary, when the response advertises one. 62 + pub binary_sha256: Option<String>, 63 +} 64 + 47 65 /// A client for one forge base URL. 48 66 #[derive(Clone, Debug)] 49 67 pub struct ForgeClient { @@ -80,30 +98,67 @@ impl ForgeClient { 80 98 json_or_error(resp).await 81 99 } 82 100 83 - /// `POST /api/v1/ci/runners/claim`. Returns `None` on 204 (nothing to run). 101 + /// `POST /api/v1/ci/runners/claim`. A 204 means nothing to run, and the reply has no job. 84 102 pub async fn claim( 85 103 &self, 86 104 runner_secret: &str, 87 105 labels: &[String], 88 - ) -> Result<Option<JobAssignment>, ClientError> { 106 + ) -> Result<ClaimReply, ClientError> { 89 107 let req = ClaimRequest { 90 108 labels: labels.to_vec(), 91 109 }; 110 + let target = format!("{}-{}", std::env::consts::ARCH, std::env::consts::OS); 92 111 let resp = self 93 112 .http 94 113 .post(self.url("/api/v1/ci/runners/claim")) 95 114 .bearer_auth(runner_secret) 115 + .header(HEADER_RUNNER_TARGET, target) 96 116 .json(&req) 97 117 .timeout(CLAIM_TIMEOUT) 98 118 .send() 99 119 .await?; 120 + let binary_sha256 = advertised_sha256(resp.headers()); 100 121 match resp.status() { 101 - StatusCode::NO_CONTENT => Ok(None), 102 - status if status.is_success() => Ok(Some(resp.json().await?)), 122 + StatusCode::NO_CONTENT => Ok(ClaimReply { 123 + job: None, 124 + binary_sha256, 125 + }), 126 + status if status.is_success() => Ok(ClaimReply { 127 + job: Some(resp.json().await?), 128 + binary_sha256, 129 + }), 103 130 _ => Err(status_error(resp).await), 104 131 } 105 132 } 106 133 134 + /// `GET /api/v1/ci/runners/binary`. Streams the forge's binary into `dest` and returns the 135 + /// sha256 (lowercase hex) of the bytes written. The caller checks it against the advertised 136 + /// value and syncs the file. 137 + pub async fn download_binary( 138 + &self, 139 + runner_secret: &str, 140 + dest: &Path, 141 + ) -> Result<String, ClientError> { 142 + let mut resp = self 143 + .http 144 + .get(self.url(BINARY_PATH)) 145 + .bearer_auth(runner_secret) 146 + .timeout(BINARY_TIMEOUT) 147 + .send() 148 + .await?; 149 + if !resp.status().is_success() { 150 + return Err(status_error(resp).await); 151 + } 152 + let mut file = tokio::fs::File::create(dest).await?; 153 + let mut hasher = Sha256::new(); 154 + while let Some(chunk) = resp.chunk().await? { 155 + hasher.update(&chunk); 156 + file.write_all(&chunk).await?; 157 + } 158 + file.flush().await?; 159 + Ok(hex::encode(hasher.finalize())) 160 + } 161 + 107 162 /// `POST /api/v1/ci/jobs/{id}/logs` with the byte offset of `body` in the job log. 108 163 pub async fn post_logs( 109 164 &self, @@ -205,6 +260,16 @@ impl ForgeClient { 205 260 } 206 261 } 207 262 263 +/// The `Forge-Binary-Sha256` header, trimmed and lowercased. An empty value counts as absent. 264 +fn advertised_sha256(headers: &HeaderMap) -> Option<String> { 265 + headers 266 + .get(HEADER_BINARY_SHA256)? 267 + .to_str() 268 + .ok() 269 + .map(|v| v.trim().to_ascii_lowercase()) 270 + .filter(|v| !v.is_empty()) 271 +} 272 + 208 273 async fn json_or_error<T: DeserializeOwned>(resp: reqwest::Response) -> Result<T, ClientError> { 209 274 if resp.status().is_success() { 210 275 Ok(resp.json().await?)
@@ -130,12 +130,21 @@ pub struct RunnerConfig { 130 130 pub sandbox: SandboxConfig, 131 131 #[serde(default, rename = "registry")] 132 132 pub registries: Vec<RegistryConfig>, 133 + /// When true (the default), an idle runner replaces itself with the forge's binary when the 134 + /// forge advertises a different one on the claim response. The runner then exits with code 135 + /// 75 and systemd restarts it. Set `auto_update = false` to pin the binary by hand. 136 + #[serde(default = "default_auto_update")] 137 + pub auto_update: bool, 133 138 } 134 139 135 140 fn default_concurrency() -> usize { 136 141 1 137 142 } 138 143 144 +fn default_auto_update() -> bool { 145 + true 146 +} 147 + 139 148 fn default_keep_days() -> u32 { 140 149 7 141 150 } @@ -159,6 +168,7 @@ impl fmt::Debug for RunnerConfig { 159 168 .field("executor", &self.executor) 160 169 .field("images", &self.images) 161 170 .field("registries", &self.registries) 171 + .field("auto_update", &self.auto_update) 162 172 .finish() 163 173 } 164 174 } @@ -268,6 +278,7 @@ mod tests { 268 278 images: ImagesConfig::default(), 269 279 sandbox: SandboxConfig::default(), 270 280 registries: Vec::new(), 281 + auto_update: true, 271 282 } 272 283 } 273 284 @@ -340,6 +351,20 @@ mod tests { 340 351 assert_eq!(config.images.keep_days, 7); 341 352 assert_eq!(config.limits.image_max_gb, 20); 342 353 assert!(config.registries.is_empty()); 354 + assert!(config.auto_update); 355 + } 356 + 357 + #[test] 358 + fn auto_update_can_be_turned_off() { 359 + let dir = tempfile::tempdir().expect("tempdir"); 360 + let path = dir.path().join("runner.toml"); 361 + std::fs::write( 362 + &path, 363 + "url = \"https://f\"\nrunner_id = \"r\"\nrunner_secret = \"s\"\nname = \"n\"\n\ 364 + work_dir = \"/w\"\nauto_update = false\n", 365 + ) 366 + .expect("write"); 367 + assert!(!load(&path).expect("load").auto_update); 343 368 } 344 369 345 370 #[test]
@@ -519,7 +519,7 @@ impl ImageSource for ImageStore { 519 519 } 520 520 521 521 /// Runs jobs in a libcontainer sandbox, started as `helper` (`forge-runner __sandbox`). It needs 522 -/// no external container tool. See `docs/ci-builtin-executor.md`. 522 +/// no external container tool. See `docs/ci.md`. 523 523 pub struct BuiltinExecutor<I = ImageStore> { 524 524 images: I, 525 525 limits: Limits,
@@ -14,4 +14,5 @@ pub mod logs; 14 14 pub mod oci; 15 15 pub mod runner; 16 16 pub mod sandbox; 17 +pub mod update; 17 18 pub mod workspace;
@@ -16,6 +16,7 @@ use crate::client::{ClientError, ForgeClient}; 16 16 use crate::config::RunnerConfig; 17 17 use crate::executor::{ExecRequest, ExecResult, Executor}; 18 18 use crate::logs::{self, PumpConfig}; 19 +use crate::update::{self, Installed, UpdateState}; 19 20 use crate::workspace; 20 21 21 22 /// Log lines queued between the executor and the log shipper. @@ -46,6 +47,15 @@ impl Default for RunOptions { 46 47 } 47 48 } 48 49 50 +/// How [`Runner::run`] ended. 51 +#[derive(Clone, Copy, Debug, PartialEq, Eq)] 52 +pub enum RunEnd { 53 + /// Shutdown was requested. 54 + Stopped, 55 + /// The binary was replaced. The caller exits with [`update::RESTART_EXIT_CODE`]. 56 + Updated, 57 +} 58 + 49 59 /// A runner bound to one forge and one executor. 50 60 #[derive(Clone)] 51 61 pub struct Runner { @@ -76,50 +86,114 @@ impl Runner { 76 86 77 87 /// Claims and runs jobs, up to `concurrency` at a time, until `shutdown` is cancelled. Jobs 78 88 /// still running at shutdown are cancelled and reported as cancelled. 79 - pub async fn run(&self, shutdown: CancellationToken) { 89 + /// 90 + /// When the forge advertises a different binary and `auto_update` is on, no new jobs are 91 + /// claimed. Once the running jobs finish, the binary is replaced and the loop returns 92 + /// [`RunEnd::Updated`]. 93 + pub async fn run(&self, shutdown: CancellationToken) -> RunEnd { 80 94 let semaphore = Arc::new(Semaphore::new(self.config.concurrency)); 81 95 let mut jobs: JoinSet<()> = JoinSet::new(); 96 + // Hashed once at startup. Without it, or with auto_update off, updates stay disabled. 97 + let exe = if self.config.auto_update { 98 + match update::locate_self() { 99 + Ok(found) => Some(found), 100 + Err(e) => { 101 + tracing::warn!(error = %e, "auto update disabled, cannot read own binary"); 102 + None 103 + } 104 + } 105 + } else { 106 + None 107 + }; 108 + let mut updates = UpdateState::new( 109 + self.config.auto_update, 110 + exe.as_ref().map(|(_, sha)| sha.clone()), 111 + ); 82 112 tracing::info!( 83 113 name = %self.config.name, 84 114 concurrency = self.config.concurrency, 115 + auto_update = updates.enabled(), 85 116 "runner started" 86 117 ); 87 - loop { 118 + let end = loop { 119 + if let Some((exe_path, _)) = &exe 120 + && let Some(sha) = updates.pending().map(str::to_owned) 121 + { 122 + if !jobs.is_empty() { 123 + // Claiming is paused until the running jobs finish. 124 + tokio::select! { 125 + _ = shutdown.cancelled() => break RunEnd::Stopped, 126 + Some(joined) = jobs.join_next() => { 127 + if let Err(e) = joined { 128 + tracing::error!(error = %e, "job task panicked"); 129 + } 130 + } 131 + } 132 + continue; 133 + } 134 + match update::install(&self.client, &self.config.runner_secret, exe_path, &sha) 135 + .await 136 + { 137 + Ok(Installed::Replaced) => { 138 + tracing::info!(sha = %sha, "binary updated, restarting"); 139 + break RunEnd::Updated; 140 + } 141 + Ok(Installed::Listed) => { 142 + tracing::warn!(sha = %sha, "forge binary is on the bad list, not updating"); 143 + updates.give_up(&sha); 144 + } 145 + Err(e) => { 146 + tracing::warn!(sha = %sha, error = %e, "update failed, keeping this binary"); 147 + updates.give_up(&sha); 148 + } 149 + } 150 + continue; 151 + } 152 + 88 153 let permit = tokio::select! { 89 - _ = shutdown.cancelled() => break, 154 + _ = shutdown.cancelled() => break RunEnd::Stopped, 90 155 permit = semaphore.clone().acquire_owned() => match permit { 91 156 Ok(permit) => permit, 92 - Err(_) => break, 157 + Err(_) => break RunEnd::Stopped, 93 158 }, 94 159 }; 95 160 let claimed = tokio::select! { 96 161 _ = shutdown.cancelled() => { 97 162 drop(permit); 98 - break; 163 + break RunEnd::Stopped; 99 164 } 100 165 r = self.client.claim(&self.config.runner_secret, &self.config.labels) => r, 101 166 }; 102 167 match claimed { 103 - Ok(Some(job)) => { 104 - tracing::info!( 105 - job_id = %job.job_id, 106 - job_name = %job.job_name, 107 - repo = %job.repo, 108 - "claimed job" 109 - ); 110 - let runner = self.clone(); 111 - let cancel = shutdown.child_token(); 112 - jobs.spawn(async move { 113 - let _permit = permit; 114 - runner.process(job, cancel).await; 115 - }); 168 + Ok(reply) => { 169 + if updates.observe(reply.binary_sha256.as_deref()) 170 + && let Some(sha) = updates.pending() 171 + { 172 + tracing::info!(sha = %sha, "forge advertises a new binary, updating when idle"); 173 + } 174 + match reply.job { 175 + Some(job) => { 176 + tracing::info!( 177 + job_id = %job.job_id, 178 + job_name = %job.job_name, 179 + repo = %job.repo, 180 + "claimed job" 181 + ); 182 + let runner = self.clone(); 183 + let cancel = shutdown.child_token(); 184 + jobs.spawn(async move { 185 + let _permit = permit; 186 + runner.process(job, cancel).await; 187 + }); 188 + } 189 + None => drop(permit), 190 + } 116 191 } 117 - Ok(None) => drop(permit), 118 192 Err(e) => { 119 193 drop(permit); 120 194 tracing::warn!(error = %e, "claim failed, backing off"); 121 195 tokio::select! { 122 - _ = shutdown.cancelled() => break, 196 + _ = shutdown.cancelled() => break RunEnd::Stopped, 123 197 _ = tokio::time::sleep(self.options.claim_backoff) => {} 124 198 } 125 199 } @@ -129,13 +203,14 @@ impl Runner { 129 203 tracing::error!(error = %e, "job task panicked"); 130 204 } 131 205 } 132 - } 206 + }; 133 207 while let Some(joined) = jobs.join_next().await { 134 208 if let Err(e) = joined { 135 209 tracing::error!(error = %e, "job task panicked"); 136 210 } 137 211 } 138 212 tracing::info!("runner stopped"); 213 + end 139 214 } 140 215 141 216 /// Runs one claimed job end to end and reports its result. Returns the final outcome.
@@ -3,7 +3,7 @@ 3 3 //! The runner never calls libcontainer itself. libcontainer forks and clones, which is unsafe in 4 4 //! a multi-threaded tokio process, so the runner spawns `forge-runner __sandbox` as a 5 5 //! single-threaded child. That child reads a [`SandboxRequest`] on stdin and runs the job. See 6 -//! `docs/ci-builtin-executor.md`. 6 +//! `docs/ci.md`. 7 7 8 8 pub mod exec; 9 9 pub mod helper;
@@ -0,0 +1,330 @@ 1 +//! Self-update for an idle runner. Each claim response may advertise the sha256 of the forge's 2 +//! binary. When that differs from ours, the runner stops claiming, lets its running jobs finish, 3 +//! swaps its own binary for the forge's, and exits with [`RESTART_EXIT_CODE`] so systemd starts 4 +//! the new one. 5 +//! 6 +//! [`UpdateState`] makes the decisions and does no I/O. [`install`] does the file work. 7 + 8 +use std::io::Read; 9 +use std::os::unix::fs::PermissionsExt; 10 +use std::path::{Path, PathBuf}; 11 +use std::process::Stdio; 12 +use std::time::Duration; 13 + 14 +use sha2::{Digest, Sha256}; 15 + 16 +use crate::client::{ClientError, ForgeClient}; 17 + 18 +/// The exit code after a successful swap. systemd restarts the runner on any non-zero code. 19 +pub const RESTART_EXIT_CODE: u8 = 75; 20 +/// How long the new binary gets to answer `--version`. 21 +const SMOKE_TIMEOUT: Duration = Duration::from_secs(10); 22 +/// The temp file, next to the running binary, that the download goes to. 23 +const NEW_NAME: &str = ".forge-runner.new"; 24 +/// The previous binary, kept for a manual rollback. 25 +const PREV_SUFFIX: &str = ".prev"; 26 +/// Optional list of bad binaries, one sha256 per line, next to the running binary. 27 +const BAD_SUFFIX: &str = ".bad"; 28 + 29 +/// Decides whether an update is due. It remembers the last sha that failed, so the runner does 30 +/// not retry it on every claim. A different advertised sha clears that memory. 31 +#[derive(Clone, Debug, PartialEq, Eq)] 32 +pub struct UpdateState { 33 + enabled: bool, 34 + /// The sha256 of the running binary. Updates are disabled when it is unknown. 35 + current: Option<String>, 36 + pending: Option<String>, 37 + given_up: Option<String>, 38 +} 39 + 40 +impl UpdateState { 41 + /// `current` is the sha256 (lowercase hex) of the running binary. 42 + pub fn new(auto_update: bool, current: Option<String>) -> Self { 43 + Self { 44 + enabled: auto_update && current.is_some(), 45 + current, 46 + pending: None, 47 + given_up: None, 48 + } 49 + } 50 + 51 + /// Records the sha the forge advertised on a claim response. Returns true when this makes a 52 + /// new update due. An absent header changes nothing. 53 + pub fn observe(&mut self, advertised: Option<&str>) -> bool { 54 + if !self.enabled { 55 + return false; 56 + } 57 + let Some(advertised) = advertised else { 58 + return false; 59 + }; 60 + if Some(advertised) == self.current.as_deref() 61 + || Some(advertised) == self.given_up.as_deref() 62 + { 63 + // Already running it, or it failed and the forge has not moved on. 64 + self.pending = None; 65 + return false; 66 + } 67 + if self.pending.as_deref() == Some(advertised) { 68 + return false; 69 + } 70 + self.pending = Some(advertised.to_owned()); 71 + true 72 + } 73 + 74 + /// Whether updates are on at all. False when `auto_update` is off or the binary is unknown. 75 + pub fn enabled(&self) -> bool { 76 + self.enabled 77 + } 78 + 79 + /// The sha to update to, if an update is due. 80 + pub fn pending(&self) -> Option<&str> { 81 + self.pending.as_deref() 82 + } 83 + 84 + /// Stops trying `sha` until the forge advertises a different one. 85 + pub fn give_up(&mut self, sha: &str) { 86 + self.given_up = Some(sha.to_owned()); 87 + self.pending = None; 88 + } 89 +} 90 + 91 +/// Whether a `.bad` list, one sha256 per line, names `sha`. 92 +pub fn is_listed_bad(list: &str, sha: &str) -> bool { 93 + list.lines() 94 + .any(|line| line.trim().eq_ignore_ascii_case(sha)) 95 +} 96 + 97 +/// The sha256 (lowercase hex) of the file at `path`. 98 +pub fn file_sha256(path: &Path) -> std::io::Result<String> { 99 + let mut file = std::fs::File::open(path)?; 100 + let mut hasher = Sha256::new(); 101 + let mut buf = vec![0u8; 64 * 1024]; 102 + loop { 103 + let n = file.read(&mut buf)?; 104 + if n == 0 { 105 + break; 106 + } 107 + hasher.update(&buf[..n]); 108 + } 109 + Ok(hex::encode(hasher.finalize())) 110 +} 111 + 112 +/// The path of the running binary, and the sha256 of its contents. 113 +pub fn locate_self() -> std::io::Result<(PathBuf, String)> { 114 + let exe = std::env::current_exe()?; 115 + let sha = file_sha256(&exe)?; 116 + Ok((exe, sha)) 117 +} 118 + 119 +/// What [`install`] did when it did not fail. 120 +#[derive(Clone, Copy, Debug, PartialEq, Eq)] 121 +pub enum Installed { 122 + /// The new binary is in place. The caller should exit with [`RESTART_EXIT_CODE`]. 123 + Replaced, 124 + /// The `.bad` list names the advertised binary, so nothing was replaced. 125 + Listed, 126 +} 127 + 128 +/// Why an update failed. The current binary is untouched in every case. 129 +#[derive(Debug, thiserror::Error)] 130 +pub enum UpdateError { 131 + #[error("local io: {0}")] 132 + Io(#[from] std::io::Error), 133 + #[error("download: {0}")] 134 + Download(#[from] ClientError), 135 + #[error("sha256 mismatch: forge advertised {expected}, downloaded {actual}")] 136 + Mismatch { expected: String, actual: String }, 137 + #[error("new binary failed its check: {0}")] 138 + Smoke(String), 139 +} 140 + 141 +/// Downloads the forge's binary next to `exe`, checks it, and swaps it in. The old binary is kept 142 +/// as `<exe>.prev`. On any error the temp file is removed. 143 +pub async fn install( 144 + client: &ForgeClient, 145 + runner_secret: &str, 146 + exe: &Path, 147 + expected: &str, 148 +) -> Result<Installed, UpdateError> { 149 + let new = exe.with_file_name(NEW_NAME); 150 + // A temp file left by an earlier crash must not be reused. 151 + let _ = std::fs::remove_file(&new); 152 + let result = stage_and_swap(client, runner_secret, exe, &new, expected).await; 153 + if !matches!(result, Ok(Installed::Replaced)) { 154 + let _ = std::fs::remove_file(&new); 155 + } 156 + result 157 +} 158 + 159 +async fn stage_and_swap( 160 + client: &ForgeClient, 161 + runner_secret: &str, 162 + exe: &Path, 163 + new: &Path, 164 + expected: &str, 165 +) -> Result<Installed, UpdateError> { 166 + let actual = client.download_binary(runner_secret, new).await?; 167 + if actual != expected { 168 + return Err(UpdateError::Mismatch { 169 + expected: expected.to_owned(), 170 + actual, 171 + }); 172 + } 173 + std::fs::set_permissions(new, std::fs::Permissions::from_mode(0o755))?; 174 + std::fs::File::open(new)?.sync_all()?; 175 + smoke_test(new).await?; 176 + 177 + let bad = match std::fs::read_to_string(sibling(exe, BAD_SUFFIX)) { 178 + Ok(list) => list, 179 + Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(), 180 + Err(e) => return Err(e.into()), 181 + }; 182 + if is_listed_bad(&bad, expected) { 183 + return Ok(Installed::Listed); 184 + } 185 + 186 + let prev = sibling(exe, PREV_SUFFIX); 187 + match std::fs::remove_file(&prev) { 188 + Ok(()) => {} 189 + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} 190 + Err(e) => return Err(e.into()), 191 + } 192 + // A hard link keeps the old inode alive at no cost. Copy only where links are not allowed. 193 + if std::fs::hard_link(exe, &prev).is_err() { 194 + std::fs::copy(exe, &prev)?; 195 + } 196 + std::fs::rename(new, exe)?; 197 + Ok(Installed::Replaced) 198 +} 199 + 200 +/// Runs `<bin> --version` and requires exit 0 within [`SMOKE_TIMEOUT`]. 201 +async fn smoke_test(bin: &Path) -> Result<(), UpdateError> { 202 + let output = tokio::time::timeout( 203 + SMOKE_TIMEOUT, 204 + tokio::process::Command::new(bin) 205 + .arg("--version") 206 + .stdin(Stdio::null()) 207 + .kill_on_drop(true) 208 + .output(), 209 + ) 210 + .await 211 + .map_err(|_| UpdateError::Smoke("--version did not finish in time".into()))??; 212 + if !output.status.success() { 213 + return Err(UpdateError::Smoke(format!( 214 + "--version exited with {}", 215 + output.status 216 + ))); 217 + } 218 + Ok(()) 219 +} 220 + 221 +/// `exe` with `suffix` appended to its file name, for example `forge.prev`. 222 +fn sibling(exe: &Path, suffix: &str) -> PathBuf { 223 + let mut name = exe.as_os_str().to_owned(); 224 + name.push(suffix); 225 + PathBuf::from(name) 226 +} 227 + 228 +#[cfg(test)] 229 +mod tests { 230 + use super::*; 231 + 232 + const OURS: &str = "aaaa"; 233 + const THEIRS: &str = "bbbb"; 234 + 235 + fn state(auto_update: bool) -> UpdateState { 236 + UpdateState::new(auto_update, Some(OURS.to_owned())) 237 + } 238 + 239 + #[test] 240 + fn a_different_sha_becomes_pending_once() { 241 + let mut s = state(true); 242 + assert!(s.observe(Some(THEIRS))); 243 + assert_eq!(s.pending(), Some(THEIRS)); 244 + // The same sha again is not news. 245 + assert!(!s.observe(Some(THEIRS))); 246 + assert_eq!(s.pending(), Some(THEIRS)); 247 + } 248 + 249 + #[test] 250 + fn matching_sha_or_no_header_leaves_nothing_pending() { 251 + let mut s = state(true); 252 + assert!(!s.observe(Some(OURS))); 253 + assert_eq!(s.pending(), None); 254 + assert!(!s.observe(None)); 255 + assert_eq!(s.pending(), None); 256 + } 257 + 258 + #[test] 259 + fn absent_header_keeps_a_pending_update() { 260 + let mut s = state(true); 261 + s.observe(Some(THEIRS)); 262 + s.observe(None); 263 + assert_eq!(s.pending(), Some(THEIRS)); 264 + } 265 + 266 + #[test] 267 + fn disabled_or_unknown_current_never_pends() { 268 + let mut off = state(false); 269 + assert!(!off.observe(Some(THEIRS))); 270 + assert_eq!(off.pending(), None); 271 + 272 + let mut unknown = UpdateState::new(true, None); 273 + assert!(!unknown.observe(Some(THEIRS))); 274 + assert_eq!(unknown.pending(), None); 275 + } 276 + 277 + #[test] 278 + fn a_failed_sha_is_not_retried_until_the_forge_moves_on() { 279 + let mut s = state(true); 280 + s.observe(Some(THEIRS)); 281 + s.give_up(THEIRS); 282 + assert_eq!(s.pending(), None); 283 + 284 + // Claims keep advertising the same sha: nothing becomes pending. 285 + assert!(!s.observe(Some(THEIRS))); 286 + assert_eq!(s.pending(), None); 287 + 288 + // A different sha is due again. 289 + assert!(s.observe(Some("cccc"))); 290 + assert_eq!(s.pending(), Some("cccc")); 291 + } 292 + 293 + #[test] 294 + fn going_back_to_the_failed_sha_cancels_a_pending_update() { 295 + let mut s = state(true); 296 + s.observe(Some(THEIRS)); 297 + s.give_up(THEIRS); 298 + s.observe(Some("cccc")); 299 + s.observe(Some(THEIRS)); 300 + assert_eq!(s.pending(), None); 301 + } 302 + 303 + #[test] 304 + fn bad_list_matches_whole_lines_case_insensitively() { 305 + let list = "# comment is not a sha\nBBBB\n cccc \n"; 306 + assert!(is_listed_bad(list, THEIRS)); 307 + assert!(is_listed_bad(list, "cccc")); 308 + assert!(!is_listed_bad(list, "bbb")); 309 + assert!(!is_listed_bad("", THEIRS)); 310 + } 311 + 312 + #[test] 313 + fn sibling_appends_the_suffix() { 314 + assert_eq!( 315 + sibling(Path::new("/home/t/.local/bin/forge"), PREV_SUFFIX), 316 + PathBuf::from("/home/t/.local/bin/forge.prev") 317 + ); 318 + } 319 + 320 + #[test] 321 + fn file_sha256_matches_a_known_digest() { 322 + let dir = tempfile::tempdir().expect("tempdir"); 323 + let path = dir.path().join("f"); 324 + std::fs::write(&path, b"abc").expect("write"); 325 + assert_eq!( 326 + file_sha256(&path).expect("hash"), 327 + "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" 328 + ); 329 + } 330 +}
@@ -16,7 +16,7 @@ use forge_ci::proto::{FinishRequest, JobAssignment, Outcome, RegisterResponse}; 16 16 use forge_runner::config::{Limits, RunnerConfig}; 17 17 use forge_runner::executor::HostExecutor; 18 18 use forge_runner::logs::PumpConfig; 19 -use forge_runner::runner::{RunOptions, Runner}; 19 +use forge_runner::runner::{RunEnd, RunOptions, Runner}; 20 20 use http_body_util::{BodyExt, Full}; 21 21 use hyper::body::Incoming; 22 22 use hyper::header::{AUTHORIZATION, HeaderMap}; @@ -321,6 +321,7 @@ fn config(url: &str, work_dir: PathBuf) -> RunnerConfig { 321 321 images: forge_runner::config::ImagesConfig::default(), 322 322 sandbox: forge_runner::config::SandboxConfig::default(), 323 323 registries: Vec::new(), 324 + auto_update: true, 324 325 } 325 326 } 326 327 @@ -349,7 +350,7 @@ async fn wait_until(what: &str, mut check: impl FnMut() -> bool) { 349 350 struct Harness { 350 351 state: Shared, 351 352 stop: CancellationToken, 352 - runner_task: tokio::task::JoinHandle<()>, 353 + runner_task: tokio::task::JoinHandle<RunEnd>, 353 354 _dir: tempfile::TempDir, 354 355 } 355 356
@@ -1,4 +1,4 @@ 1 -//! Backup sets (phase 1, see `docs/backup-interfaces.md`). 1 +//! Backup sets (phase 1, see `docs/backup.md`). 2 2 //! 3 3 //! A set is a directory `forge-backup-<UTC stamp>/` holding a consistent DB snapshot, one git 4 4 //! bundle per non-empty repo, optionally the instance key, and a `manifest.json` with a sha256 for
@@ -55,6 +55,13 @@ pub async fn serve(config: Config) -> forge_core::Result<()> { 55 55 let forge_bin = std::env::current_exe()?.canonicalize()?; 56 56 forge_git::setup::write_hook_shims(&data_dir, &forge_bin)?; 57 57 58 + // Hash the running binary once, before any runner can ask for it. Hashing reads the whole 59 + // executable, which takes tens of seconds for a debug build, so it runs off the boot path and 60 + // the listeners bind first. Any caller that needs the hash waits on the same `OnceLock`. 61 + tokio::task::spawn_blocking(|| { 62 + crate::self_update::running(); 63 + }); 64 + 58 65 let internal_token: Arc<str> = Arc::from(random_hex()); 59 66 let admin_token: Arc<str> = Arc::from(random_hex()); 60 67 crate::admin_client::write_admin_token(&data_dir, &admin_token)?; @@ -99,6 +106,8 @@ pub async fn serve(config: Config) -> forge_core::Result<()> { 99 106 ci_waker: Arc::new(Notify::new()), 100 107 mailer, 101 108 mail_waker: Arc::new(Notify::new()), 109 + notify_hub: Arc::new(crate::notify_hub::NotifyHub::new()), 110 + ci_live: Arc::new(crate::ci::CiLive::default()), 102 111 }; 103 112 if let Err(e) = crate::signatures::rebuild_allowed_signers(&state).await { 104 113 tracing::error!(error = %e, "building the SSH allowed-signers file failed"); @@ -150,6 +159,14 @@ pub async fn serve(config: Config) -> forge_core::Result<()> { 150 159 151 160 let (stop_tx, stop_rx) = watch::channel(false); 152 161 162 + // Open notification streams never end on their own, so they are closed with the rest. 163 + let hub = Arc::clone(&state.notify_hub); 164 + let hub_stop = stop_rx.clone(); 165 + tokio::spawn(async move { 166 + wait_stop(hub_stop).await; 167 + hub.close(); 168 + }); 169 + 153 170 let ssh_config = SshConfig { 154 171 listen: ssh_listen, 155 172 host_key_path: data_dir.join("ssh_host_ed25519_key"), @@ -169,8 +186,10 @@ pub async fn serve(config: Config) -> forge_core::Result<()> { 169 186 170 187 let signal_stop = stop_tx.clone(); 171 188 tokio::spawn(async move { 172 - shutdown_signal().await; 173 - tracing::info!("shutdown signal received"); 189 + tokio::select! { 190 + () = shutdown_signal() => tracing::info!("shutdown signal received"), 191 + () = crate::self_update::restart_requested() => tracing::info!("self-update: restarting"), 192 + } 174 193 signal_stop.send_replace(true); 175 194 }); 176 195
@@ -1,4 +1,4 @@ 1 -//! Runner and job API under `/api/v1/ci` (contract: docs/ci-interfaces.md, section 4). 1 +//! Runner and job API under `/api/v1/ci` (see docs/ci.md, Runner protocol). 2 2 //! 3 3 //! Runners authenticate with `Authorization: Bearer <runner secret>`. A job's routes take 4 4 //! `Authorization: Bearer <job token>`, which `job_by_token` accepts only while the job is Running. @@ -25,6 +25,7 @@ use topcoat::router::{Body, HeaderMap, StatusCode, path_param, route}; 25 25 use crate::api::{self, Refusal}; 26 26 use crate::ci::{artifact_key, log_path}; 27 27 use crate::pages::repo::https_clone_url; 28 +use crate::self_update; 28 29 use crate::state::AppState; 29 30 30 31 path_param!(job_id); @@ -170,6 +171,13 @@ async fn register(cx: &Cx, body: Body) -> topcoat::Result<Response> { 170 171 171 172 #[route(POST "/api/v1/ci/runners/claim")] 172 173 async 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 + 180 +async fn claim_work(cx: &Cx) -> topcoat::Result<Response> { 173 181 let state = app_context::<AppState>(cx); 174 182 // The labels in the request are ignored: the labels stored at registration are used. 175 183 let runner = match runner_from(state, headers(cx)).await { @@ -177,24 +185,48 @@ async fn claim(cx: &Cx) -> topcoat::Result<Response> { 177 185 Err(res) => return Ok(*res), 178 186 }; 179 187 180 - // Register for the wake-up before the first attempt, so a job queued in between is not missed. 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. 181 191 let waker = state.ci_waker.clone(); 182 - let mut woken = std::pin::pin!(waker.notified()); 183 - woken.as_mut().enable(); 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 +} 184 210 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), 211 +/// The running binary, for a runner that updates itself. 404 unless self-update is enabled. 212 +#[route(GET "/api/v1/ci/runners/binary")] 213 +async 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); 189 217 } 190 - tokio::select! { 191 - _ = woken.as_mut() => {} 192 - _ = tokio::time::sleep(CLAIM_WAIT) => {} 218 + if !state.config.self_update.enabled { 219 + return Ok(api::error(StatusCode::NOT_FOUND, "self-update is disabled")); 193 220 } 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), 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 + } 198 230 } 199 231 } 200 232 @@ -217,6 +249,8 @@ async fn try_claim(state: &AppState, runner: &CiRunner) -> Result<Option<JobAssi 217 249 else { 218 250 return Ok(None); 219 251 }; 252 + state.ci_live.bump_job(&job.id); 253 + state.ci_live.bump_run(&run.id); 220 254 let spec: JobSpec = serde_json::from_str(&job.spec_json).map_err(api::internal)?; 221 255 let repo = state 222 256 .store @@ -370,6 +404,7 @@ async fn write_log( 370 404 .add_log_bytes(&job.id, len as i64) 371 405 .await 372 406 .map_err(api::internal)?; 407 + state.ci_live.bump_job(&job.id); 373 408 Ok(LogWrite::Stored(size + len)) 374 409 } 375 410 @@ -591,7 +626,8 @@ async fn finish(cx: &Cx, body: Body) -> topcoat::Result<Response> { 591 626 } 592 627 593 628 /// 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. 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. 595 631 async fn finish_job( 596 632 state: &AppState, 597 633 job: &CiJob, @@ -614,6 +650,7 @@ async fn finish_job( 614 650 .map_err(api::internal)?; 615 651 616 652 let mut queued = false; 653 + let mut any_changed = false; 617 654 for other in state 618 655 .store 619 656 .jobs_for_run(&job.run_id) @@ -621,6 +658,14 @@ async fn finish_job( 621 658 .map_err(api::internal)? 622 659 { 623 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 + } 624 669 if other.id == job.id 625 670 || (changed && matches!(other.state, JobState::Skipped | JobState::Queued)) 626 671 { @@ -628,9 +673,14 @@ async fn finish_job( 628 673 crate::ci::sync_status(state, &run, &other).await; 629 674 } 630 675 } 676 + if any_changed { 677 + state.ci_live.bump_run(&run.id); 678 + } 631 679 if queued { 632 680 state.ci_waker.notify_waiters(); 633 681 } 682 + // A run that just went green may deploy the forge itself. 683 + self_update::maybe_deploy(state, &run).await; 634 684 Ok(run.state.as_str()) 635 685 } 636 686
@@ -769,3 +769,133 @@ async fn job_token_reads_over_smart_http_but_never_pushes() { 769 769 "a finished job's token must stop working" 770 770 ); 771 771 } 772 + 773 +/// Like `bare`, with self-update enabled for `OWNER/REPO`. 774 +async fn bare_with_self_update() -> Bare { 775 + let dir = tempfile::tempdir().expect("tempdir"); 776 + let data_dir = dir.path().to_path_buf(); 777 + let mut state = test_state(&data_dir).await; 778 + let mut config = (*state.config).clone(); 779 + config.self_update.enabled = true; 780 + config.self_update.repo = format!("{OWNER}/{REPO}"); 781 + state.config = Arc::new(config); 782 + let store = state.store.clone(); 783 + let router = app_router(state); 784 + let owner = store 785 + .create_user(OWNER, None, "not-a-real-hash", false) 786 + .await 787 + .unwrap(); 788 + let repo = store 789 + .create_repo(&owner.id, REPO, "", Visibility::Public, "main") 790 + .await 791 + .unwrap(); 792 + Bare { 793 + router, 794 + store, 795 + data_dir, 796 + owner_id: owner.id, 797 + repo, 798 + _dir: dir, 799 + } 800 +} 801 + 802 +#[tokio::test] 803 +async fn runner_binary_needs_runner_auth_and_self_update() { 804 + let env = bare(|_| {}).await; 805 + let (_, secret) = runner(&env, &[]).await; 806 + let uri = "/api/v1/ci/runners/binary"; 807 + 808 + let sent = send(&env.router, "GET", uri, &[], Vec::new()).await; 809 + assert_eq!(sent.status, 401); 810 + // Self-update is off in this fixture, so an authorized runner gets 404. 811 + let sent = send(&env.router, "GET", uri, &[bearer(&secret)], Vec::new()).await; 812 + assert_eq!(sent.status, 404); 813 +} 814 + 815 +#[tokio::test] 816 +async fn runner_binary_streams_the_running_binary_with_its_sha() { 817 + use sha2::{Digest, Sha256}; 818 + 819 + let env = bare_with_self_update().await; 820 + let (_, secret) = runner(&env, &[]).await; 821 + let uri = "/api/v1/ci/runners/binary"; 822 + 823 + let sent = send( 824 + &env.router, 825 + "GET", 826 + uri, 827 + &[bearer("not-a-secret")], 828 + Vec::new(), 829 + ) 830 + .await; 831 + assert_eq!(sent.status, 401); 832 + 833 + let sent = send(&env.router, "GET", uri, &[bearer(&secret)], Vec::new()).await; 834 + assert_eq!(sent.status, 200); 835 + assert_eq!( 836 + sent.headers.get(http::header::CONTENT_TYPE).unwrap(), 837 + "application/octet-stream" 838 + ); 839 + let advertised = sent 840 + .headers 841 + .get("forge-binary-sha256") 842 + .unwrap() 843 + .to_str() 844 + .unwrap(); 845 + let digest = hex::encode(Sha256::digest(&sent.body)); 846 + assert_eq!(digest, advertised); 847 + assert_eq!( 848 + advertised, 849 + crate::self_update::running() 850 + .expect("hashed at startup") 851 + .sha256 852 + ); 853 +} 854 + 855 +#[tokio::test] 856 +async fn claim_advertises_the_binary_only_to_a_matching_target() { 857 + let env = bare_with_self_update().await; 858 + let (_, secret) = runner(&env, &["linux"]).await; 859 + let target = crate::self_update::target(); 860 + let claim_with = |target: Option<String>| { 861 + let mut headers = vec![bearer(&secret)]; 862 + if let Some(target) = target { 863 + headers.push(("Forge-Runner-Target", target)); 864 + } 865 + headers 866 + }; 867 + 868 + // A 204 (no work) carries the header for a matching target. 869 + let sent = send( 870 + &env.router, 871 + "POST", 872 + "/api/v1/ci/runners/claim", 873 + &claim_with(Some(target.clone())), 874 + Vec::new(), 875 + ) 876 + .await; 877 + assert_eq!(sent.status, 204); 878 + let running = crate::self_update::running().expect("hashed at startup"); 879 + assert_eq!( 880 + sent.headers 881 + .get("forge-binary-sha256") 882 + .unwrap() 883 + .to_str() 884 + .unwrap(), 885 + running.sha256 886 + ); 887 + 888 + // No target, or another target: no header. 889 + for requested in [None, Some("riscv64-plan9".to_owned())] { 890 + let sent = send( 891 + &env.router, 892 + "POST", 893 + "/api/v1/ci/runners/claim", 894 + &claim_with(requested), 895 + Vec::new(), 896 + ) 897 + .await; 898 + assert_eq!(sent.status, 204); 899 + assert!(sent.headers.get("forge-binary-sha256").is_none()); 900 + } 901 +}
@@ -11,9 +11,11 @@ use forge_core::store::commit_status::StatusState; 11 11 use crate::state::AppState; 12 12 13 13 pub mod api; 14 +pub mod live; 14 15 pub mod reaper; 15 16 pub mod schedule; 16 17 18 +pub use live::CiLive; 17 19 pub use schedule::{manual_run, on_pull_event, on_ref_update}; 18 20 19 21 /// Longest status description written, in characters. Longer text is cut.
@@ -1,5 +1,4 @@ 1 -//! Git LFS over HTTP: the batch API, object GET and PUT, and verify. See 2 -//! `docs/wave-b-interfaces.md`, section 2. 1 +//! Git LFS over HTTP: the batch API, object GET and PUT, and verify. 3 2 //! 4 3 //! Objects are content-addressed under `<data_dir>/lfs/<oid[0..2]>/<oid[2..4]>/<oid>` and shared by 5 4 //! every repo. The `lfs_objects` table says which repo may use which object. Uploads stream into
@@ -29,12 +29,14 @@ mod lfs_pointer; 29 29 mod mail; 30 30 mod mirrors; 31 31 mod notify; 32 +mod notify_hub; 32 33 mod pages; 33 34 mod post_update; 34 35 mod pulls_sync; 35 36 mod raw; 36 37 mod render; 37 38 mod replication; 39 +mod self_update; 38 40 mod signatures; 39 41 mod ssh_backend; 40 42 mod state; @@ -137,6 +139,14 @@ async fn real_main() -> ExitCode { 137 139 138 140 let outcome: Result<(), BoxError> = match parts.as_slice() { 139 141 [] | ["serve"] => serve(&config_path).await, 142 + ["--version"] => { 143 + println!( 144 + "forge {} {}", 145 + env!("CARGO_PKG_VERSION"), 146 + env!("FORGE_BUILD_SHA") 147 + ); 148 + return ExitCode::SUCCESS; 149 + } 140 150 ["hook", stage] => hook_cmd::run(stage).await.map_err(Into::into), 141 151 ["admin", "create-user", username] => admin_cli::create_user(&config_path, username, admin) 142 152 .await @@ -209,6 +219,8 @@ async fn real_main() -> ExitCode { 209 219 }; 210 220 211 221 match outcome { 222 + // A self-update swap asks for a restart. systemd Restart=on-failure starts the new binary. 223 + Ok(()) if self_update::restart_pending() => ExitCode::from(self_update::RESTART_EXIT_CODE), 212 224 Ok(()) => ExitCode::SUCCESS, 213 225 Err(e) => { 214 226 eprintln!("forge: {e}");
@@ -1,5 +1,5 @@ 1 1 //! Page: source archive at `/{owner}/{repo}/archive/{ref}.tar.gz` or `.zip`. The `git archive` 2 -//! stdout is streamed as the response body (see docs/m2-interfaces.md). 2 +//! stdout is streamed as the response body. 3 3 4 4 use std::process::Stdio; 5 5
@@ -2,7 +2,6 @@ 2 2 //! from raw paths, since no ref name may start with `-`. It reads the Git LFS pointer at 3 3 //! `ref:path` and streams the object it names, when the repo links that object. Access follows 4 4 //! the raw route in `raw.rs`: public repos are open, and private repos need a signed `?token=`. 5 -//! See `docs/wave-b-interfaces.md`, section 2. 6 5 7 6 use forge_core::raw_token::verify_raw; 8 7 use forge_core::{Repo, Visibility};
@@ -815,6 +815,7 @@ mod tests { 815 815 advisories: Default::default(), 816 816 mail: Default::default(), 817 817 orgs: Default::default(), 818 + self_update: Default::default(), 818 819 }; 819 820 let state = AppState { 820 821 config: Arc::new(config), @@ -836,6 +837,8 @@ mod tests { 836 837 ci_waker: Arc::new(tokio::sync::Notify::new()), 837 838 mailer: Arc::new(crate::mail::FakeMailer::default()), 838 839 mail_waker: Arc::new(tokio::sync::Notify::new()), 840 + notify_hub: Arc::new(crate::notify_hub::NotifyHub::new()), 841 + ci_live: Arc::new(crate::ci::CiLive::default()), 839 842 }; 840 843 let hash = hash_password_async(PASSWORD.into()).await.unwrap(); 841 844 let user = store
@@ -1,5 +1,5 @@ 1 1 //! Raw user-content origin (separate listener, explicit routes; never `.discover()`). 2 -//! See docs/m2-interfaces.md, "Raw origin". 2 +//! See docs/architecture.md, section 5.5. 3 3 4 4 use std::path::Path; 5 5 use std::process::Stdio;
@@ -1,4 +1,4 @@ 1 -//! syntect class-based highlighting (see docs/m2-interfaces.md). 1 +//! syntect class-based highlighting. 2 2 //! 3 3 //! Output is class-based (`hl-` prefix) and never carries inline styles, so the 4 4 //! CSP without 'unsafe-inline' is respected. Each source line becomes one HTML
@@ -1,4 +1,4 @@ 1 -//! comrak + ammonia markdown rendering (see docs/m2-interfaces.md). 1 +//! comrak + ammonia markdown rendering. 2 2 //! 3 3 //! Pipeline: parse with comrak (GFM extensions, raw HTML passed through), 4 4 //! optionally rewrite link and image URLs in the AST, render to HTML, then
@@ -0,0 +1,575 @@ 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 + 10 +use std::path::{Path, PathBuf}; 11 +use std::sync::OnceLock; 12 +use std::sync::atomic::{AtomicBool, Ordering}; 13 +use std::time::Duration; 14 + 15 +use forge_core::ci::{CiRun, RunState}; 16 +use forge_core::config::SelfUpdateConfig; 17 +use futures_util::StreamExt; 18 +use futures_util::stream::BoxStream; 19 +use http::header::{CONTENT_LENGTH, CONTENT_TYPE, HeaderValue}; 20 +use http_body_util::StreamBody; 21 +use sha2::{Digest, Sha256}; 22 +use tokio::io::AsyncWriteExt; 23 +use tokio_util::io::ReaderStream; 24 +use topcoat::router::{Body, HeaderMap, StatusCode, response::Response}; 25 + 26 +use crate::state::AppState; 27 + 28 +/// Exit code after a swap. systemd `Restart=on-failure` restarts the unit on any non-zero exit. 29 +pub const RESTART_EXIT_CODE: u8 = 75; 30 +/// Response header carrying the sha256 of the running binary, lowercase hex. 31 +pub const SHA_HEADER: &str = "Forge-Binary-Sha256"; 32 +/// Request header a runner sends with its `<arch>-<os>` target. 33 +pub const TARGET_HEADER: &str = "Forge-Runner-Target"; 34 +/// How long the staged binary may take to answer `--version`. 35 +const SANITY_TIMEOUT: Duration = Duration::from_secs(10); 36 + 37 +/// Serialises deploys. A deploy that finds another one running is skipped. 38 +static DEPLOY_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); 39 +/// Wakes the server so it stops after a swap. 40 +static RESTART: tokio::sync::Notify = tokio::sync::Notify::const_new(); 41 +static RESTART_REQUESTED: AtomicBool = AtomicBool::new(false); 42 + 43 +/// The running binary, hashed once at startup. 44 +pub 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 + 50 +static 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. 54 +pub 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 + 69 +fn 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. 77 +fn 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. 87 +pub async fn restart_requested() { 88 + RESTART.notified().await; 89 +} 90 + 91 +/// Whether a swap asked the process to exit with `RESTART_EXIT_CODE`. 92 +pub 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`. 97 +pub 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. 102 +fn 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. 109 +pub 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. 126 +pub 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`. 139 +fn 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. 148 +pub 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. 184 +pub 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 + 194 +async 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. 266 +async 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. 290 +async 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`. 311 +fn 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. 324 +pub 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. 345 +pub 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 + 365 +fn 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)] 379 +mod 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 +}
@@ -41,4 +41,8 @@ pub struct AppState { 41 41 pub mailer: Arc<dyn crate::mail::Mailer>, 42 42 /// Wakes the mail worker when a message is queued or the mail settings change. 43 43 pub mail_waker: Arc<Notify>, 44 + /// Open notification streams per user. A change to an inbox publishes the new unread count. 45 + pub notify_hub: Arc<crate::notify_hub::NotifyHub>, 46 + /// Change counters of CI jobs and runs. Log followers and run pages wait on them. 47 + pub ci_live: Arc<crate::ci::CiLive>, 44 48 }
@@ -51,6 +51,7 @@ pub fn test_config(dir: &Path) -> Config { 51 51 advisories: Default::default(), 52 52 mail: Default::default(), 53 53 orgs: Default::default(), 54 + self_update: Default::default(), 54 55 } 55 56 } 56 57 @@ -78,6 +79,8 @@ pub async fn test_state(dir: &Path) -> AppState { 78 79 ci_waker: Arc::new(Notify::new()), 79 80 mailer: Arc::new(crate::mail::FakeMailer::default()), 80 81 mail_waker: Arc::new(Notify::new()), 82 + notify_hub: Arc::new(crate::notify_hub::NotifyHub::new()), 83 + ci_live: Arc::new(crate::ci::CiLive::default()), 81 84 } 82 85 } 83 86
@@ -1,7 +1,7 @@ 1 1 //! JSON payload builders for webhook deliveries. Each builder returns the event-specific fields. 2 2 //! `emit` adds `action` (when the event has one), `repository` and `sender`. 3 3 //! 4 -//! The shapes follow docs/webhooks-interfaces.md. Links are built from `base_url` (the configured 4 +//! Links are built from `base_url` (the configured 5 5 //! public URL), so they point at the instance a receiver can reach. 6 6 7 7 use forge_core::{Comment, Issue, PullRequest, Repo, Review, Visibility};
@@ -420,6 +420,7 @@ impl Worker { 420 420 .claim(&self.secret, &self.labels) 421 421 .await 422 422 .expect("claim") 423 + .job 423 424 } 424 425 425 426 /// Claims and runs jobs until the forge has nothing left. Returns how many ran. @@ -471,6 +472,8 @@ async fn register_runner(b: &Booted, admin_session: &str, name: &str) -> Worker 471 472 images: forge_runner::config::ImagesConfig::default(), 472 473 sandbox: forge_runner::config::SandboxConfig::default(), 473 474 registries: Vec::new(), 475 + // The e2e runner must not replace the test binary under itself. 476 + auto_update: false, 474 477 }; 475 478 let runner = 476 479 Runner::new(config, Arc::new(HostExecutor), RunOptions::default()).expect("runner");
@@ -0,0 +1,27 @@ 1 +#!/bin/sh 2 +# Roll back ~/bin/forge to ~/bin/forge.prev after a unit failed. 3 +# Usage: forge-rollback <unit-name>. Output goes to the journal. 4 +set -eu 5 + 6 +unit="${1:?usage: forge-rollback <unit-name>}" 7 +bin="$HOME/bin/forge" 8 +prev="$bin.prev" 9 +bad="$bin.bad" 10 + 11 +if [ ! -e "$prev" ]; then 12 + echo "forge-rollback: $unit failed and $prev does not exist, nothing to roll back" 13 + exit 0 14 +fi 15 + 16 +sha=$(sha256sum "$bin" | cut -d' ' -f1) 17 +# Keep one hash per line, even if the file was edited by hand without a final newline. 18 +if [ -s "$bad" ] && [ -n "$(tail -c 1 "$bad")" ]; then 19 + echo >> "$bad" 20 +fi 21 +echo "$sha" >> "$bad" 22 +mv -f "$prev" "$bin" 23 +echo "forge-rollback: $unit failed, marked $sha bad, restored $prev" 24 + 25 +systemctl --user reset-failed "$unit" 26 +systemctl --user start "$unit" 27 +echo "forge-rollback: started $unit on the previous binary"
@@ -0,0 +1,6 @@ 1 +[Unit] 2 +Description=roll back forge after forge.service failed 3 + 4 +[Service] 5 +Type=oneshot 6 +ExecStart=%h/bin/forge-rollback forge.service
@@ -0,0 +1,6 @@ 1 +[Unit] 2 +Description=roll back forge runner after forge-runner.service failed 3 + 4 +[Service] 5 +Type=oneshot 6 +ExecStart=%h/bin/forge-rollback forge-runner.service
@@ -0,0 +1,18 @@ 1 +[Unit] 2 +Description=forge CI runner 3 +After=network-online.target 4 +Wants=network-online.target 5 +# Crash loops stop the unit: 3 starts within 60s, then it enters failed and OnFailure runs the rollback. 6 +StartLimitIntervalSec=60 7 +StartLimitBurst=3 8 +OnFailure=forge-runner-rollback.service 9 + 10 +[Service] 11 +ExecStart=%h/bin/forge runner run --config %h/.config/forge-runner/runner.toml 12 +# A self-update exits with code 75, which is non-zero, so Restart=on-failure restarts it into the new binary. 13 +Restart=on-failure 14 +RestartSec=10 15 +TimeoutStopSec=120 16 + 17 +[Install] 18 +WantedBy=default.target
@@ -0,0 +1,20 @@ 1 +[Unit] 2 +Description=forge 3 +After=network-online.target 4 +# Crash loops stop the unit: 3 starts within 60s, then it enters failed and OnFailure runs the rollback. 5 +StartLimitIntervalSec=60 6 +StartLimitBurst=3 7 +OnFailure=forge-rollback.service 8 + 9 +[Service] 10 +WorkingDirectory=%h 11 +Environment=FORGE_CONFIG=%h/config/forge.toml 12 +ExecStart=%h/bin/forge serve 13 +# A self-update exits with code 75, which is non-zero, so Restart=on-failure restarts it into the new binary. 14 +Restart=on-failure 15 +RestartSec=2 16 +NoNewPrivileges=yes 17 +LimitNOFILE=65536 18 + 19 +[Install] 20 +WantedBy=default.target