flagon-io/g1t

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Merge the desktop app's runner: it starts with the app for the signed-in person, sandboxed by default and running nothing where no sandbox is available yet, registers through the site, keeps itself up to date, and says everywhere what it is doing, why it is not, and how to fix it.yesterday		1	//! The runner in the app: g1t's self-hosted runner, started for the
		2	//! signed-in person when g1t opens and stopped when it quits, so their
		3	//! computer runs their workspace's `runs-on: self-hosted` jobs with
		4	//! nothing to install, register or start by hand
		5	//! (docs.g1t.sh/guides/desktop/#the-runner).
		6	//!
		7	//! - **When.** On by default; settings turn it off ("Run jobs on this
		8	//! computer while g1t is open"), and Stop in settings or the tray stops
		9	//! it until Start, or until g1t starts again. It starts a few seconds
		10	//! after the window, never before.
		11	//! - **How.** The backend settings choose (backend.rs): Sandboxed, the
		12	//! default, where nothing a job does reaches this computer; Docker; or
		13	//! directly on this computer. The two that are not sandboxes are only
		14	//! ever chosen by the person, and nothing falls back to them.
		15	//! - **Registering.** The first time, and when the workspace or backend
		16	//! changes, the app needs a registration token for the workspace. Only
		17	//! the site can make one, for an owner who is signed in: the app asks
		18	//! the site's page (`g1t-runner-wanted`), the page asks the server
		19	//! (`POST /-/desktop/runner-token`) and hands the token to the app
		20	//! (`runner_register` in bridge.rs), which gives it to `g1t-runner` in
		21	//! its environment, once. It is never shown or written to the log. The
		22	//! registration is kept (registration.json, and the runner's own folder
		23	//! or volume) and used on every later start.
		24	//! - **Up to date.** Soon after it starts and every six hours, while the
		25	//! runner is idle, the app looks for a newer release, or a newer image,
		26	//! and puts it in place: never in the middle of a job.
		27	//! - **Seen.** What it is doing, and when it is not running why and how to
		28	//! fix it (diagnosis.rs), is in the tray, in settings, and on the site's
		29	//! Runners page (`g1t-runner-status`), read from what its log says
		30	//! (log.rs). The whole log is kept in `runner/runner.log`.
		31	//!
		32	//! All of it runs on a thread of its own, which owns the runner's process
		33	//! and is told what happened through one channel.
		34
		35	mod backend;
		36	mod container;
		37	pub mod diagnosis;
		38	mod launcher;
		39	mod log;
		40	mod native;
		41
		42	use std::collections::VecDeque;
		43	use std::fs::{File, OpenOptions};
		44	use std::io::{BufRead, BufReader, Read, Write};
		45	use std::path::PathBuf;
		46	use std::process::Child;
		47	use std::sync::Mutex;
		48	use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
		49	use std::time::{Duration, Instant};
		50
		51	use serde::{Deserialize, Serialize};
		52	use tauri::{AppHandle, Emitter, EventTarget, Manager};
		53	use url::Url;
		54
		55	pub use backend::Backend;
		56	use backend::{Engine, Missing};
		57	use diagnosis::{Code, Diagnosis, Facts};
		58
		59	use crate::{config, frame, settings, tray};
		60
		61	/// What settings, the tray and the site listen for: the runner's status changed.
		62	pub const STATUS_EVENT: &str = "g1t-runner-status";
		63	/// What the site's page listens for: the app needs a registration token for a workspace.
		64	pub const WANTED_EVENT: &str = "g1t-runner-wanted";
		65
		66	/// How long after the window the runner starts.
		67	const AFTER_START: Duration = Duration::from_secs(4);
		68	/// How often the thread looks at the runner when nothing has happened.
		69	const TICK: Duration = Duration::from_secs(5);
		70	/// How long after starting the first look for an update is, then how often.
		71	const FIRST_UPDATE: Duration = Duration::from_secs(90);
		72	const UPDATE_EVERY: Duration = Duration::from_secs(6 * 60 * 60);
		73	/// How often a backend that cannot start is looked at again.
		74	const LOOK_AGAIN: Duration = Duration::from_secs(30);
		75	/// How long the app waits for the site to hand over a token before asking again.
		76	const ASK_AGAIN: Duration = Duration::from_secs(60);
		77	/// The longest the app waits before starting a runner that stopped by itself again.
		78	const MOST_BACKOFF: Duration = Duration::from_secs(10 * 60);
		79	/// The runner says when a poll fails, and nothing when one works: this
		80	/// long without another failure, it is back (its longest wait is 60s).
		81	const BACK_ONLINE: Duration = Duration::from_secs(75);
		82	/// The runner's log is started over when it is bigger than this.
		83	const LOG_LIMIT: u64 = 5 * 1024 * 1024;
		84	/// The last lines of the log settings show.
		85	const TAIL: usize = 40;
		86
		87	/// A workspace the site says the person may add runners to: those they own.
		88	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
		89	pub struct Offered {
		90	pub slug: String,
		91	#[serde(default)]
		92	pub name: Option<String>,
		93	}
		94
		95	/// What the runner is doing, as the thread last said.
		96	#[derive(Clone, Debug, Default, PartialEq)]
		97	struct Now {
		98	code: Code,
		99	detail: Option<String>,
		100	exit_code: Option<i32>,
		101	wait_seconds: Option<u64>,
		102	job: Option<String>,
		103	version: Option<String>,
		104	}
		105
		106	/// What the site last said about the person.
		107	#[derive(Clone, Debug, Default, PartialEq)]
		108	struct Site {
		109	/// Unknown until the site's first page loads.
		110	signed_in: Option<bool>,
		111	owned: Vec<Offered>,
		112	current: Option<String>,
		113	}
		114
		115	/// What the rest of the app shares with the thread.
		116	#[derive(Default)]
		117	pub struct Shared {
		118	now: Mutex<Now>,
		119	/// The workspace a token is wanted for.
		120	wanted: Mutex<Option<String>>,
		121	site: Mutex<Site>,
		122	/// Stopped by the person until Start, or until g1t starts again.
		123	paused: Mutex<bool>,
		124	tail: Mutex<VecDeque<String>>,
		125	tx: Mutex<Option<Sender<Ask>>>,
		126	}
		127
		128	/// What settings, the tray and the site are shown.
		129	#[derive(Clone, Debug, PartialEq, Serialize)]
		130	#[serde(rename_all = "camelCase")]
		131	pub struct View {
		132	pub diagnosis: Diagnosis,
		133	/// What the site says of this computer once g1t is closed on it.
		134	pub closed: Diagnosis,
		135	pub running: bool,
		136	pub version: Option<String>,
		137	pub backend: Backend,
		138	pub workspace: Option<String>,
		139	/// The workspaces it may run for, from the site.
		140	pub workspaces: Vec<Offered>,
		141	/// The setting: run jobs while g1t is open.
		142	pub enabled: bool,
		143	pub paused: bool,
		144	/// The workspace's Runners page.
		145	pub link: Option<String>,
		146	/// This computer's name, which the runner has on g1t.
		147	pub machine: String,
		148	pub platform: String,
		149	/// The last lines of the runner's log.
		150	pub log: Vec<String>,
		151	}
		152
		153	/// What the thread is told.
		154	enum Ask {
		155	/// Something changed: settings, Start or Stop, Retry.
		156	Look,
		157	Retry,
		158	Token { workspace: String, token: String },
		159	Refused { workspace: String, allowed: bool, why: String },
		160	Line(u64, String),
		161	Exited(u64),
		162	Quit(Sender<()>),
		163	}
		164
		165	/// The registration the app made, besides what the runner keeps itself.
		166	#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
		167	struct Registration {
		168	server: String,
		169	workspace: String,
		170	backend: Backend,
		171	}
		172
		173	fn folder(app: &AppHandle) -> PathBuf {
		174	app.path().app_data_dir().unwrap_or_else(|_| std::env::temp_dir().join("sh.g1t.desktop")).join("runner")
		175	}
		176
		177	impl Registration {
		178	fn path(app: &AppHandle) -> PathBuf {
		179	folder(app).join("registration.json")
		180	}
		181	fn load(app: &AppHandle) -> Option<Registration> {
		182	std::fs::read(Self::path(app)).ok().and_then(|bytes| serde_json::from_slice(&bytes).ok())
		183	}
		184	fn save(&self, app: &AppHandle) {
		185	let _ = std::fs::create_dir_all(folder(app));
		186	let _ = std::fs::write(Self::path(app), serde_json::to_vec_pretty(self).unwrap_or_default());
		187	}
		188	fn clear(app: &AppHandle) {
		189	let _ = std::fs::remove_file(Self::path(app));
		190	}
		191	}
		192
		193	/// Where a native runner of a backend keeps its configuration.
		194	fn native_dir(app: &AppHandle, backend: Backend) -> PathBuf {
		195	folder(app).join(match backend {
		196	Backend::Docker => "docker",
		197	Backend::Host => "host",
		198	Backend::Sandboxed => launcher::FOLDER,
		199	})
		200	}
		201
		202	/// The workspace the runner is for: the one settings name, else the one it
		203	/// is registered with, else the one the person is in on the site, else the
		204	/// first they own.
		205	fn workspace(app: &AppHandle) -> Option<String> {
		206	let settings = config::settings(app);
		207	if let Some(chosen) = settings.runner_workspace.filter(|w| !w.is_empty()) {
		208	return Some(chosen);
		209	}
		210	let server = config::server(app).to_string();
		211	if let Some(registration) = Registration::load(app).filter(|r| r.server == server) {
		212	return Some(registration.workspace);
		213	}
		214	let site = app.state::<Shared>().site.lock().unwrap().clone();
		215	site.current.filter(|slug| site.owned.iter().any(|o| &o.slug == slug)).or_else(|| site.owned.first().map(|o| o.slug.clone()))
		216	}
		217
		218	fn facts(app: &AppHandle, now: &Now) -> Facts {
		219	let server = config::server(app);
		220	Facts {
		221	workspace: workspace(app),
		222	backend: config::settings(app).runner_backend.label(),
		223	machine: backend::machine_name(),
		224	job: now.job.clone(),
		225	detail: now.detail.clone(),
		226	exit_code: now.exit_code,
		227	wait_seconds: now.wait_seconds,
		228	server: server.host_str().unwrap_or("g1t").to_owned(),
		229	platform: std::env::consts::OS.into(),
		230	}
		231	}
		232
		233	pub fn view(app: &AppHandle) -> View {
		234	let shared = app.state::<Shared>();
		235	let now = shared.now.lock().unwrap().clone();
		236	let settings = config::settings(app);
		237	let facts = facts(app, &now);
		238	let server = config::server(app);
		239	View {
		240	diagnosis: diagnosis::diagnose(now.code, &facts),
		241	closed: diagnosis::closed(&facts),
		242	running: matches!(now.code, Code::Idle | Code::Busy | Code::Offline),
		243	version: now.version,
		244	backend: settings.runner_backend,
		245	link: facts.workspace.as_ref().and_then(|w| server.join(&format!("{w}/-/runners")).ok()).map(|u| u.to_string()),
		246	workspace: facts.workspace,
		247	workspaces: shared.site.lock().unwrap().owned.clone(),
		248	enabled: settings.runner,
		249	paused: *shared.paused.lock().unwrap(),
		250	machine: facts.machine,
		251	platform: facts.platform,
		252	log: shared.tail.lock().unwrap().iter().cloned().collect(),
		253	}
		254	}
		255
		256	/// The tray's line: `Runner: Waiting for jobs`, at most 60 characters.
		257	pub fn headline(view: &View) -> String {
		258	let line = format!("Runner: {}", view.diagnosis.title);
		259	if line.chars().count() > 60 { line.chars().take(59).collect::<String>() + "…" } else { line }
		260	}
		261
		262	/// Tells everyone what the runner is doing now.
		263	fn publish(app: &AppHandle) {
		264	let view = view(app);
		265	tray::runner_changed(app, &view);
		266	let _ = app.emit_to(settings::SETTINGS, STATUS_EVENT, view.clone());
		267	let _ = app.emit_to(EventTarget::webview(frame::SITE), STATUS_EVENT, view);
		268	}
		269
		270	fn ask(app: &AppHandle, ask: Ask) {
		271	if let Some(tx) = app.state::<Shared>().tx.lock().unwrap().as_ref() {
		272	let _ = tx.send(ask);
		273	}
		274	}
		275
		276	/// Starts the thread that keeps the runner, as the app starts.
		277	pub fn install(app: &AppHandle) {
		278	let (tx, rx) = mpsc::channel();
		279	*app.state::<Shared>().tx.lock().unwrap() = Some(tx.clone());
		280	let handle = app.clone();
		281	let spawned = std::thread::Builder::new().name("g1t-runner".into()).spawn(move || Keeper::new(handle, tx).keep(rx));
		282	if let Err(error) = spawned {
		283	eprintln!("g1t: the runner's thread could not start: {error}");
		284	}
		285	}
		286
		287	/// Settings changed: the thread looks again.
		288	pub fn settings_changed(app: &AppHandle) {
		289	ask(app, Ask::Look);
		290	publish(app);
		291	}
		292
		293	/// Start or Stop, from settings or the tray.
		294	pub fn set_paused(app: &AppHandle, paused: bool) {
		295	*app.state::<Shared>().paused.lock().unwrap() = paused;
		296	// Start with the setting off turns the setting on: that is what was asked.
		297	if !paused && !config::settings(app).runner {
		298	let mut settings = config::settings(app);
		299	settings.runner = true;
		300	let _ = settings::apply(app, settings);
		301	}
		302	settings_changed(app);
		303	}
		304
		305	/// The tray's Start or Stop row.
		306	pub fn toggle(app: &AppHandle) {
		307	let view = view(app);
		308	set_paused(app, view.enabled && !view.paused);
		309	}
		310
		311	/// Retry and Restart: now, rather than when the wait is over.
		312	pub fn retry(app: &AppHandle) {
		313	*app.state::<Shared>().paused.lock().unwrap() = false;
		314	ask(app, Ask::Retry);
		315	}
		316
		317	/// What the site says about the person, on every page: whether they are
		318	/// signed in, the workspaces they own, and the one they are in.
		319	pub fn offer(app: &AppHandle, signed_in: bool, owned: Vec<Offered>, current: Option<String>) {
		320	let next = Site { signed_in: Some(signed_in), owned, current: current.map(|c| c.to_ascii_lowercase()) };
		321	let changed = {
		322	let shared = app.state::<Shared>();
		323	let mut site = shared.site.lock().unwrap();
		324	let changed = *site != next;
		325	*site = next;
		326	changed
		327	};
		328	if changed {
		329	settings_changed(app);
		330	}
		331	}
		332
		333	/// The site handed over a registration token, or said why it could not:
		334	/// `allowed` is false when the person may not add runners to the workspace.
		335	pub fn register(app: &AppHandle, workspace: String, token: Option<String>, allowed: bool, error: Option<String>) {
		336	match token {
		337	Some(token) if token.starts_with("g1trt_") => ask(app, Ask::Token { workspace, token }),
		338	_ => ask(app, Ask::Refused { workspace, allowed, why: error.unwrap_or_else(|| "g1t didn't make a registration token.".into()) }),
		339	}
		340	}
		341
		342	/// A diagnosis's button, or Start and Stop: from the settings window, or
		343	/// from the site's card for this computer.
		344	pub fn act(app: &AppHandle, action: &str) -> Result<(), String> {
		345	use tauri_plugin_opener::OpenerExt;
		346	let browse = |url: &str| app.opener().open_url(url, None::<&str>).map_err(|e| e.to_string());
		347	match action {
		348	"start" => {
		349	set_paused(app, false);
		350	retry(app);
		351	}
		352	"stop" => set_paused(app, true),
		353	"retry" | "restart" => retry(app),
		354	"sign_in" => crate::links::go(app, "/login"),
		355	"new_workspace" => crate::links::go(app, "/workspaces/new"),
		356	"view_runners" => match workspace(app) {
		357	Some(workspace) => crate::links::go(app, &format!("/{workspace}/-/runners")),
		358	None => crate::links::go(app, "/"),
		359	},
		360	"choose_workspace" | "choose_backend" => settings::open(app),
		361	"view_log" => {
		362	let path = log_file(app);
		363	if !path.exists() {
		364	return Err("The runner hasn't written a log yet.".into());
		365	}
		366	app.opener().open_path(path.to_string_lossy(), None::<&str>).map_err(|e| e.to_string())?;
		367	}
		368	"open_docker" => open_docker(app)?,
		369	"get_docker" => browse("https://docs.docker.com/get-started/get-docker/")?,
		370	"install_sysbox" => browse("https://github.com/nestybox/sysbox#installation")?,
		371	"learn_more" => browse("https://docs.g1t.sh/guides/desktop/#the-runner")?,
		372	"report" => browse(config::server(app).join("support").map_err(|e| e.to_string())?.as_str())?,
		373	other => return Err(format!("the runner has no action called {other}")),
		374	}
		375	Ok(())
		376	}
		377
		378	/// Starts Docker Desktop, where it is installed in its usual place.
		379	fn open_docker(app: &AppHandle) -> Result<(), String> {
		380	use tauri_plugin_opener::OpenerExt;
		381	if cfg!(target_os = "macos") {
		382	return app.opener().open_path("/Applications/Docker.app", None::<&str>).map_err(|e| e.to_string());
		383	}
		384	if cfg!(windows) {
		385	let program = std::env::var_os("ProgramFiles").map(PathBuf::from).unwrap_or_else(|| PathBuf::from(r"C:\Program Files"));
		386	let desktop = program.join("Docker").join("Docker").join("Docker Desktop.exe");
		387	if desktop.exists() {
		388	return app.opener().open_path(desktop.to_string_lossy(), None::<&str>).map_err(|e| e.to_string());
		389	}
		390	return Err("Docker Desktop isn't installed in its usual place. Get Docker to install it.".into());
		391	}
		392	Err("Start Docker with your system's service manager, such as `sudo systemctl start docker`.".into())
		393	}
		394
		395	/// The workspace a token is wanted for now, if one is: the site asks as it loads.
		396	pub fn wanted(app: &AppHandle) -> Option<String> {
		397	app.state::<Shared>().wanted.lock().unwrap().clone()
		398	}
		399
		400	/// The runner's log, for the settings window to open.
		401	pub fn log_file(app: &AppHandle) -> PathBuf {
		402	folder(app).join("runner.log")
		403	}
		404
		405	/// Stops the runner as g1t quits or restarts, and waits for it.
		406	pub fn shutdown(app: &AppHandle) {
		407	let tx = app.state::<Shared>().tx.lock().unwrap().take();
		408	if let Some(tx) = tx {
		409	let (done, wait) = mpsc::channel();
		410	if tx.send(Ask::Quit(done)).is_ok() {
		411	let _ = wait.recv_timeout(Duration::from_secs(30));
		412	}
		413	}
		414	}
		415
		416	/// How the runner is started this time.
		417	enum Process {
		418	/// `g1t-runner` or its sandbox launcher, a child of the app.
		419	Child(Child),
		420	/// The runner's container, and `docker logs --follow` on it.
		421	Container(Child),
		422	}
		423
		424	struct Run {
		425	generation: u64,
		426	engine: Engine,
		427	backend: Backend,
		428	workspace: String,
		429	process: Process,
		430	/// Registered and polling.
		431	listening: bool,
		432	busy: bool,
		433	/// When a poll last failed to reach g1t.
		434	offline_at: Option<Instant>,
		435	/// Started with a registration token, which a container keeps in its configuration until replaced.
		436	with_token: bool,
		437	forgotten: bool,
		438	token_refused: bool,
		439	error: Option<String>,
		440	}
		441
		442	struct Keeper {
		443	app: AppHandle,
		444	tx: Sender<Ask>,
		445	run: Option<Run>,
		446	generation: u64,
		447	token: Option<(String, String)>,
		448	not_before: Instant,
		449	failures: u32,
		450	next_update: Instant,
		451	log: Option<File>,
		452	}
		453
		454	/// What a set of the state carries besides its code.
		455	#[derive(Default)]
		456	struct Extra {
		457	detail: Option<String>,
		458	exit_code: Option<i32>,
		459	wait_seconds: Option<u64>,
		460	}
		461
		462	fn detail(text: impl Into<String>) -> Extra {
		463	Extra { detail: Some(text.into()), ..Extra::default() }
		464	}
		465
		466	impl Keeper {
		467	fn new(app: AppHandle, tx: Sender<Ask>) -> Keeper {
		468	let now = Instant::now();
		469	Keeper { app, tx, run: None, generation: 0, token: None, not_before: now + AFTER_START, failures: 0, next_update: now + FIRST_UPDATE, log: None }
		470	}
		471
		472	fn keep(mut self, rx: Receiver<Ask>) {
		473	self.reap();
		474	self.set(Code::Starting, detail("Starting after g1t's window."));
		475	loop {
		476	match rx.recv_timeout(TICK) {
		477	Ok(Ask::Quit(done)) => {
		478	self.halt();
		479	let _ = done.send(());
		480	return;
		481	}
		482	Ok(ask) => self.handle(ask),
		483	Err(RecvTimeoutError::Timeout) => {}
		484	Err(RecvTimeoutError::Disconnected) => {
		485	self.halt();
		486	return;
		487	}
		488	}
		489	self.look();
		490	}
		491	}
		492
		493	fn set(&mut self, code: Code, extra: Extra) {
		494	let changed = {
		495	let shared = self.app.state::<Shared>();
		496	let mut now = shared.now.lock().unwrap();
		497	let job = if code == Code::Busy { now.job.clone() } else { None };
		498	let next = Now { code, detail: extra.detail, exit_code: extra.exit_code, wait_seconds: extra.wait_seconds, job, version: now.version.clone() };
		499	let changed = *now != next;
		500	*now = next;
		501	changed
		502	};
		503	if changed {
		504	publish(&self.app);
		505	}
		506	}
		507
		508	fn code(&self) -> Code {
		509	self.app.state::<Shared>().now.lock().unwrap().code
		510	}
		511
		512	fn note(&mut self, job: Option<String>, version: Option<String>) {
		513	let shared = self.app.state::<Shared>();
		514	let mut now = shared.now.lock().unwrap();
		515	if job.is_some() {
		516	now.job = job;
		517	}
		518	if version.is_some() {
		519	now.version = version;
		520	}
		521	}
		522
		523	/// A line of the runner's log, into the log file and the tail settings show.
		524	fn write_log(&mut self, line: &str) {
		525	{
		526	let shared = self.app.state::<Shared>();
		527	let mut tail = shared.tail.lock().unwrap();
		528	if tail.len() == TAIL {
		529	tail.pop_front();
		530	}
		531	tail.push_back(line.to_owned());
		532	}
		533	let path = log_file(&self.app);
		534	if self.log.is_none() {
		535	let _ = std::fs::create_dir_all(folder(&self.app));
		536	if std::fs::metadata(&path).is_ok_and(|m| m.len() > LOG_LIMIT) {
		537	let _ = std::fs::rename(&path, path.with_extension("log.old"));
		538	}
		539	self.log = OpenOptions::new().create(true).append(true).open(&path).ok();
		540	}
		541	if let Some(file) = self.log.as_mut() {
		542	let _ = writeln!(file, "{line}");
		543	}
		544	}
		545
		546	fn handle(&mut self, ask: Ask) {
		547	match ask {
		548	Ask::Look | Ask::Quit(_) => {}
		549	Ask::Retry => {
		550	self.not_before = Instant::now();
		551	self.failures = 0;
		552	match self.code() {
		553	// The runner is fine: only the update is tried again.
		554	Code::UpdateFailed => self.next_update = Instant::now(),
		555	Code::Offline => self.halt(),
		556	_ if self.run.as_ref().is_some_and(|run| !run.listening) => self.halt(),
		557	_ => {}
		558	}
		559	}
		560	Ask::Token { workspace, token } => {
		561	let shared = self.app.state::<Shared>();
		562	let mut wanted = shared.wanted.lock().unwrap();
		563	if wanted.as_deref() == Some(workspace.as_str()) {
		564	*wanted = None;
		565	self.token = Some((workspace, token));
		566	self.not_before = Instant::now();
		567	}
		568	}
		569	Ask::Refused { workspace, allowed, why } => {
		570	let matched = {
		571	let shared = self.app.state::<Shared>();
		572	let mut wanted = shared.wanted.lock().unwrap();
		573	let matched = wanted.as_deref() == Some(workspace.as_str());
		574	if matched {
		575	*wanted = None;
		576	}
		577	matched
		578	};
		579	if matched {
		580	self.not_before = Instant::now() + MOST_BACKOFF;
		581	self.set(if allowed { Code::RegisterFailed } else { Code::NotAllowed }, detail(why));
		582	}
		583	}
		584	Ask::Line(generation, line) => self.line(generation, &line),
		585	Ask::Exited(generation) => self.exited(generation),
		586	}
		587	}
		588
		589	fn line(&mut self, generation: u64, line: &str) {
		590	let Some(run) = self.run.as_mut().filter(|run| run.generation == generation) else { return };
		591	let mut became = None;
		592	let mut job = None;
		593	let mut version = None;
		594	match log::read(line) {
		595	log::Event::Listening { version: v } => {
		596	run.listening = true;
		597	version = Some(v);
		598	became = Some(Code::Idle);
		599	}
		600	log::Event::Took { name, repo } => {
		601	run.busy = true;
		602	run.offline_at = None;
		603	job = Some(format!("{name} ({repo})"));
		604	became = Some(Code::Busy);
		605	}
		606	log::Event::Ended => {
		607	run.busy = false;
		608	became = Some(Code::Idle);
		609	}
		610	log::Event::Offline => {
		611	run.offline_at = Some(Instant::now());
		612	if !run.busy {
		613	became = Some(Code::Offline);
		614	}
		615	}
		616	log::Event::Forgotten => run.forgotten = true,
		617	log::Event::TokenRefused => run.token_refused = true,
		618	log::Event::Failed(error) => run.error = Some(error),
		619	log::Event::Other => {}
		620	}
		621	let registered = version.is_some().then(|| Registration {
		622	server: config::server(&self.app).to_string(),
		623	workspace: run.workspace.clone(),
		624	backend: run.backend,
		625	});
		626	// A container started with a token keeps it in its configuration:
		627	// it is replaced at once by one without, which needs no token.
		628	let replace = registered.is_some() && run.with_token && matches!(run.process, Process::Container(_));
		629	self.write_log(line);
		630	if let Some(registration) = registered {
		631	registration.save(&self.app);
		632	self.failures = 0;
		633	}
		634	self.note(job, version);
		635	if let Some(code) = became {
		636	self.set(code, Extra::default());
		637	}
		638	if replace {
		639	self.halt();
		640	self.not_before = Instant::now();
		641	}
		642	}
		643
		644	fn exited(&mut self, generation: u64) {
		645	if self.run.as_ref().is_none_or(|run| run.generation != generation) {
		646	return;
		647	}
		648	let mut run = self.run.take().expect("checked above");
		649	let exit_code = match &mut run.process {
		650	Process::Child(child) => child.wait().ok().and_then(|status| status.code()),
		651	Process::Container(_) => container::exit_code(),
		652	};
		653	self.stop(&mut run);
		654	if run.forgotten || run.token_refused {
		655	// The registration is no more, or was never made: the next start asks for a token.
		656	self.forget();
		657	self.not_before = Instant::now();
		658	return;
		659	}
		660	self.failures += 1;
		661	let wait = Duration::from_secs(15 * 2u64.pow(self.failures.min(6))).min(MOST_BACKOFF);
		662	self.not_before = Instant::now() + wait;
		663	let offline = run.error.as_deref().is_some_and(|e| e.contains("could not reach g1t"));
		664	if offline {
		665	self.set(Code::Offline, Extra::default());
		666	} else if !run.listening && Registration::load(&self.app).is_none() {
		667	self.set(Code::RegisterFailed, Extra { detail: run.error, ..Extra::default() });
		668	} else {
		669	self.set(Code::Crashed, Extra { detail: run.error, exit_code, wait_seconds: Some(wait.as_secs()) });
		670	}
		671	}
		672
		673	/// Stops a run's process, the way it was started.
		674	fn stop(&self, run: &mut Run) {
		675	match (&mut run.process, run.engine) {
		676	(Process::Container(logs), _) => container::stop(Some(logs)),
		677	(Process::Child(child), Engine::Launcher) => {
		678	launcher::down(&native::binary(&folder(&self.app)), &native_dir(&self.app, run.backend));
		679	native::stop(child);
		680	}
		681	// The image's runner is always a container; a child here is only ever g1t-runner's.
		682	(Process::Child(child), Engine::Sysbox) => native::stop(child),
		683	(Process::Child(child), Engine::Native { docker }) => {
		684	native::stop(child);
		685	// A job's container outlives the runner killed under it: removed now, not at its next start.
		686	if docker {
		687	native::sweep(&native_dir(&self.app, run.backend));
		688	}
		689	}
		690	}
		691	let _ = std::fs::remove_file(folder(&self.app).join("pid"));
		692	}
		693
		694	/// Stops whatever runs now.
		695	fn halt(&mut self) {
		696	self.generation += 1;
		697	if let Some(mut run) = self.run.take() {
		698	self.stop(&mut run);
		699	}
		700	}
		701
		702	/// A runner an earlier g1t started and did not stop: it ended without quitting.
		703	fn reap(&self) {
		704	if let Some(pid) = std::fs::read_to_string(folder(&self.app).join("pid")).ok().and_then(|text| text.trim().parse().ok()) {
		705	native::reap(pid);
		706	let _ = std::fs::remove_file(folder(&self.app).join("pid"));
		707	}
		708	}
		709
		710	/// Unregisters the runner the app registered, whatever backend it was.
		711	fn forget(&mut self) {
		712	let app = self.app.clone();
		713	let Some(registration) = Registration::load(&app) else { return };
		714	let binary = native::binary(&folder(&app));
		715	match registration.backend {
		716	Backend::Sandboxed if launcher::READY => {
		717	launcher::down(&binary, &native_dir(&app, Backend::Sandboxed));
		718	let _ = std::fs::remove_dir_all(native_dir(&app, Backend::Sandboxed));
		719	}
		720	Backend::Sandboxed => {
		721	if backend::docker_ready() {
		722	container::forget();
		723	}
		724	}
		725	backend => native::forget(&binary, &native_dir(&app, backend)),
		726	}
		727	Registration::clear(&app);
		728	}
		729
		730	/// Whether what the runner keeps of a registration is still here.
		731	fn kept(&self, engine: Engine, backend: Backend) -> bool {
		732	match engine {
		733	Engine::Sysbox => container::registered(),
		734	Engine::Launcher => native_dir(&self.app, backend).exists(),
		735	Engine::Native { .. } => native_dir(&self.app, backend).join("config.json").exists(),
		736	}
		737	}
		738
		739	/// Asks the site for a registration token for `workspace`.
		740	fn want(&mut self, workspace: &str) {
		741	*self.app.state::<Shared>().wanted.lock().unwrap() = Some(workspace.to_owned());
		742	let _ = self.app.emit_to(EventTarget::webview(frame::SITE), WANTED_EVENT, workspace.to_owned());
		743	self.not_before = Instant::now() + ASK_AGAIN;
		744	let signed_out = self.app.state::<Shared>().site.lock().unwrap().signed_in == Some(false);
		745	self.set(if signed_out { Code::SignedOut } else { Code::Registering }, Extra::default());
		746	}
		747
		748	/// The thread's look round: what should run, and whether it does.
		749	fn look(&mut self) {
		750	let settings = config::settings(&self.app);
		751	let paused = *self.app.state::<Shared>().paused.lock().unwrap();
		752	if !settings.runner || paused {
		753	if self.run.is_some() {
		754	self.halt();
		755	}
		756	*self.app.state::<Shared>().wanted.lock().unwrap() = None;
		757	self.set(if paused { Code::Stopped } else { Code::Off }, Extra::default());
		758	return;
		759	}
		760	let workspace = workspace(&self.app);
		761	if let Some(run) = &self.run {
		762	let moved = run.backend != settings.runner_backend || Some(&run.workspace) != workspace.as_ref();
		763	let back = run.offline_at.is_some_and(|at| at.elapsed() >= BACK_ONLINE);
		764	let (busy, listening) = (run.busy, run.listening);
		765	if back {
		766	if let Some(run) = self.run.as_mut() {
		767	run.offline_at = None;
		768	}
		769	self.set(Code::Idle, Extra::default());
		770	}
		771	// A change of backend or workspace waits for the job to end, as an update does.
		772	if moved && !busy {
		773	self.halt();
		774	self.not_before = Instant::now();
		775	} else if !busy && listening && Instant::now() >= self.next_update {
		776	self.update();
		777	}
		778	return;
		779	}
		780	if Instant::now() < self.not_before {
		781	return;
		782	}
		783	self.start(settings.runner_backend, workspace);
		784	}
		785
		786	fn start(&mut self, backend: Backend, workspace: Option<String>) {
		787	let app = self.app.clone();
		788	let engine = match backend::engine(backend) {
		789	Ok(engine) => engine,
		790	Err(missing) => {
		791	self.not_before = Instant::now() + LOOK_AGAIN;
		792	let code = match missing {
		793	Missing::Docker => Code::NoDocker,
		794	Missing::Sandbox => Code::NoSandbox,
		795	Missing::Sysbox => Code::NoSysbox,
		796	};
		797	self.set(code, Extra::default());
		798	return;
		799	}
		800	};
		801	let Some(workspace) = workspace else {
		802	self.not_before = Instant::now() + LOOK_AGAIN;
		803	let code = match app.state::<Shared>().site.lock().unwrap().signed_in {
		804	Some(false) => Code::SignedOut,
		805	Some(true) => Code::NoWorkspace,
		806	None => Code::Starting,
		807	};
		808	self.set(code, detail("Waiting for g1t to open."));
		809	return;
		810	};
		811	let server = config::server(&app);
		812	let registration = Registration::load(&app);
		813	let ours = registration.as_ref().is_some_and(|r| r.server == server.as_str() && r.workspace == workspace && r.backend == backend) && self.kept(engine, backend);
		814	let mut token = None;
		815	if !ours {
		816	if registration.is_some() {
		817	self.set(Code::Starting, detail("Removing this computer's earlier registration."));
		818	self.forget();
		819	}
		820	match self.token.take() {
		821	Some((for_workspace, given)) if for_workspace == workspace => token = Some(given),
		822	_ => return self.want(&workspace),
		823	}
		824	}
		825	// The setting names the workspace from now on, so browsing another one never moves the runner.
		826	if config::settings(&app).runner_workspace.as_deref() != Some(workspace.as_str()) {
		827	let mut settings = config::settings(&app);
		828	settings.runner_workspace = Some(workspace.clone());
		829	let _ = config::set_settings(&app, settings);
		830	}
		831	self.set(Code::Starting, Extra::default());
		832	let envs = environment(&server, backend, token.as_deref());
		833	let generation = self.generation + 1;
		834	self.generation = generation;
		835	let started = match engine {
		836	Engine::Sysbox => {
		837	if !container::have_image() {
		838	self.set(Code::Updating, detail("Downloading the runner's image."));
		839	container::pull();
		840	}
		841	container::start(&envs).map(Process::Container)
		842	}
		843	Engine::Launcher | Engine::Native { .. } => self.ensure_binary(&server).and_then(|binary| {
		844	let dir = native_dir(&app, backend);
		845	let child = if engine == Engine::Launcher {
		846	let mut command = launcher::up(&binary, &dir);
		847	command.envs(envs.iter().map(|(name, value)| (*name, value.as_str())));
		848	command.spawn()
		849	} else {
		850	native::spawn(&binary, &dir, &envs)
		851	};
		852	let child = child.map_err(|e| format!("g1t-runner could not start: {e}"))?;
		853	let _ = std::fs::write(folder(&app).join("pid"), child.id().to_string());
		854	Ok(Process::Child(child))
		855	}),
		856	};
		857	match started {
		858	Ok(mut process) => {
		859	let child = match &mut process {
		860	Process::Child(child) | Process::Container(child) => child,
		861	};
		862	self.follow(generation, child);
		863	self.run = Some(Run {
		864	generation,
		865	engine,
		866	backend,
		867	workspace,
		868	process,
		869	listening: false,
		870	busy: false,
		871	offline_at: None,
		872	with_token: token.is_some(),
		873	forgotten: false,
		874	token_refused: false,
		875	error: None,
		876	});
		877	self.set(Code::Starting, Extra::default());
		878	}
		879	Err(why) => {
		880	// The token is kept for the next try, while it lasts.
		881	if let Some(given) = token {
		882	self.token = Some((workspace, given));
		883	}
		884	self.failures += 1;
		885	self.not_before = Instant::now() + LOOK_AGAIN;
		886	let offline = why.contains("error sending request") || why.contains("dns") || why.contains("connect");
		887	self.write_log(&format!("g1t: the runner could not start: {why}"));
		888	if offline {
		889	self.set(Code::Offline, Extra::default());
		890	} else {
		891	self.set(Code::Crashed, Extra { detail: Some(why), wait_seconds: Some(LOOK_AGAIN.as_secs()), ..Extra::default() });
		892	}
		893	}
		894	}
		895	}
		896
		897	/// Reads the child's output, a line at a time, onto the thread's channel.
		898	fn follow(&self, generation: u64, child: &mut Child) {
		899	fn pipe(tx: Sender<Ask>, generation: u64, from: impl Read + Send + 'static, last: bool) {
		900	std::thread::spawn(move || {
		901	for line in BufReader::new(from).lines().map_while(Result::ok) {
		902	if tx.send(Ask::Line(generation, line)).is_err() {
		903	return;
		904	}
		905	}
		906	if last {
		907	let _ = tx.send(Ask::Exited(generation));
		908	}
		909	});
		910	}
		911	if let Some(err) = child.stderr.take() {
		912	pipe(self.tx.clone(), generation, err, false);
		913	}
		914	if let Some(out) = child.stdout.take() {
		915	pipe(self.tx.clone(), generation, out, true);
		916	}
		917	}
		918
		919	/// `g1t-runner` installed, from the newest release when it is not.
		920	fn ensure_binary(&mut self, server: &Url) -> Result<PathBuf, String> {
		921	let folder = folder(&self.app);
		922	let binary = native::binary(&folder);
		923	if native::installed(&folder).is_some() {
		924	return Ok(binary);
		925	}
		926	self.set(Code::Updating, detail("Downloading g1t-runner."));
		927	let manifest = native::latest(server)?;
		928	let bytes = native::download(server, &manifest)?;
		929	native::put(&folder, &bytes, &manifest.version)?;
		930	self.note(None, Some(manifest.version));
		931	Ok(binary)
		932	}
		933
		934	/// A newer release or image, put in place while the runner is idle.
		935	/// The new one is downloaded and checked before the runner stops, so
		936	/// one that fails leaves it running as it was.
		937	fn update(&mut self) {
		938	self.next_update = Instant::now() + UPDATE_EVERY;
		939	if !config::settings(&self.app).auto_update {
		940	return;
		941	}
		942	let Some(engine) = self.run.as_ref().map(|run| run.engine) else { return };
		943	match engine {
		944	Engine::Sysbox => {
		945	if container::pull() || container::outdated() {
		946	self.set(Code::Updating, detail("Starting the newest runner image."));
		947	self.halt();
		948	self.not_before = Instant::now();
		949	}
		950	}
		951	Engine::Launcher | Engine::Native { .. } => {
		952	let server = config::server(&self.app);
		953	let folder = folder(&self.app);
		954	let Ok(manifest) = native::latest(&server) else { return };
		955	if !native::installed(&folder).is_none_or(|have| native::newer(&manifest.version, &have)) {
		956	return;
		957	}
		958	let bytes = match native::download(&server, &manifest) {
		959	Ok(bytes) => bytes,
		960	Err(why) => {
		961	let why = format!("g1t-runner {} couldn't be installed: {why}", manifest.version);
		962	self.write_log(&format!("g1t: {why}"));
		963	return self.set(Code::UpdateFailed, detail(why));
		964	}
		965	};
		966	self.set(Code::Updating, detail(format!("Updating g1t-runner to {}.", manifest.version)));
		967	self.halt();
		968	match native::put(&folder, &bytes, &manifest.version) {
		969	Ok(()) if engine == Engine::Launcher => {
		970	let backend = config::settings(&self.app).runner_backend;
		971	launcher::update(&native::binary(&folder), &native_dir(&self.app, backend));
		972	}
		973	Ok(()) => {}
		974	Err(why) => self.write_log(&format!("g1t: g1t-runner {} couldn't be put in place: {why}", manifest.version)),
		975	}
		976	self.not_before = Instant::now();
		977	}
		978	}
		979	}
		980	}
		981
		982	/// What `register-and-run` reads: where g1t is, the runner's name and
		983	/// labels, and the token, the first time only. The app updates the runner,
		984	/// so the runner does not update itself; registering again takes the
		985	/// place of this computer's earlier runner of the same name.
		986	fn environment(server: &Url, backend: Backend, token: Option<&str>) -> Vec<(&'static str, String)> {
		987	let mut envs = vec![
		988	("G1T_RUNNER_URL", server.as_str().trim_end_matches('/').to_owned()),
		989	("G1T_RUNNER_NAME", backend::machine_name()),
		990	("G1T_RUNNER_LABELS", "desktop".to_owned()),
		991	("G1T_RUNNER_REPLACE", "1".to_owned()),
		992	("G1T_RUNNER_NO_AUTO_UPDATE", "1".to_owned()),
		993	];
		994	if backend == Backend::Host {
		995	envs.push(("G1T_RUNNER_NO_DOCKER", "1".to_owned()));
		996	}
		997	if let Some(token) = token {
		998	envs.push(("G1T_RUNNER_TOKEN", token.to_owned()));
		999	}
		1000	envs
		1001	}
		1002
		1003	#[cfg(test)]
		1004	mod tests {
		1005	use super::*;
		1006
		1007	fn view(code: Code, job: Option<&str>) -> View {
		1008	let facts = Facts { workspace: Some("acme".into()), backend: "sandboxed", machine: "box".into(), job: job.map(str::to_owned), ..Facts::default() };
		1009	View {
		1010	diagnosis: diagnosis::diagnose(code, &facts),
		1011	closed: diagnosis::closed(&facts),
		1012	running: code == Code::Idle,
		1013	version: None,
		1014	backend: Backend::Sandboxed,
		1015	workspace: Some("acme".into()),
		1016	workspaces: vec![],
		1017	enabled: true,
		1018	paused: false,
		1019	link: None,
		1020	machine: "box".into(),
		1021	platform: "windows".into(),
		1022	log: vec![],
		1023	}
		1024	}
		1025
		1026	#[test]
		1027	fn the_tray_says_what_the_runner_is_doing() {
		1028	assert_eq!(headline(&view(Code::Idle, None)), "Runner: Waiting for jobs");
		1029	assert_eq!(headline(&view(Code::Busy, Some("Build (acme/web)"))), "Runner: Running Build (acme/web)");
		1030	assert_eq!(headline(&view(Code::NoDocker, None)), "Runner: Not running: Docker isn't running");
		1031	assert_eq!(headline(&view(Code::Busy, Some(&"x".repeat(200)))).chars().count(), 60);
		1032	}
		1033
		1034	#[test]
		1035	fn the_runner_is_told_where_and_who_and_the_token_only_once() {
		1036	let server = Url::parse("https://g1t.sh/").unwrap();
		1037	let envs = environment(&server, Backend::Host, Some("g1trt_abc"));
		1038	let get = |name: &str| envs.iter().find(|(n, _)| *n == name).map(|(_, v)| v.as_str());
		1039	assert_eq!(get("G1T_RUNNER_URL"), Some("https://g1t.sh"));
		1040	assert_eq!(get("G1T_RUNNER_LABELS"), Some("desktop"));
		1041	assert_eq!(get("G1T_RUNNER_NO_DOCKER"), Some("1"));
		1042	assert_eq!(get("G1T_RUNNER_NO_AUTO_UPDATE"), Some("1"));
		1043	assert_eq!(get("G1T_RUNNER_TOKEN"), Some("g1trt_abc"));
		1044	let again = environment(&server, Backend::Docker, None);
		1045	assert!(!again.iter().any(|(name, _)| *name == "G1T_RUNNER_TOKEN" || *name == "G1T_RUNNER_NO_DOCKER"));
		1046	}
		1047
		1048	#[test]
		1049	fn the_site_names_workspaces_in_its_own_words() {
		1050	let offered: Vec<Offered> = serde_json::from_str(r#"[{"slug":"acme","name":"Acme"},{"slug":"solo"}]"#).unwrap();
		1051	assert_eq!(offered[1], Offered { slug: "solo".into(), name: None });
		1052	}
		1053	}

This file's history is long; its oldest lines are credited to the oldest commit read.