Skip to content
Sign up

flagon-io/g1t

public
0
1,053 linesCodeBlameRaw
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
35mod backend;
36mod container;
37pub mod diagnosis;
38mod launcher;
39mod log;
40mod native;
41
42use std::collections::VecDeque;
43use std::fs::{File, OpenOptions};
44use std::io::{BufRead, BufReader, Read, Write};
45use std::path::PathBuf;
46use std::process::Child;
47use std::sync::Mutex;
48use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
49use std::time::{Duration, Instant};
50
51use serde::{Deserialize, Serialize};
52use tauri::{AppHandle, Emitter, EventTarget, Manager};
53use url::Url;
54
55pub use backend::Backend;
56use backend::{Engine, Missing};
57use diagnosis::{Code, Diagnosis, Facts};
58
59use crate::{config, frame, settings, tray};
60
61/// What settings, the tray and the site listen for: the runner's status changed.
62pub const STATUS_EVENT: &str = "g1t-runner-status";
63/// What the site's page listens for: the app needs a registration token for a workspace.
64pub const WANTED_EVENT: &str = "g1t-runner-wanted";
65
66/// How long after the window the runner starts.
67const AFTER_START: Duration = Duration::from_secs(4);
68/// How often the thread looks at the runner when nothing has happened.
69const TICK: Duration = Duration::from_secs(5);
70/// How long after starting the first look for an update is, then how often.
71const FIRST_UPDATE: Duration = Duration::from_secs(90);
72const UPDATE_EVERY: Duration = Duration::from_secs(6 * 60 * 60);
73/// How often a backend that cannot start is looked at again.
74const LOOK_AGAIN: Duration = Duration::from_secs(30);
75/// How long the app waits for the site to hand over a token before asking again.
76const ASK_AGAIN: Duration = Duration::from_secs(60);
77/// The longest the app waits before starting a runner that stopped by itself again.
78const 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).
81const BACK_ONLINE: Duration = Duration::from_secs(75);
82/// The runner's log is started over when it is bigger than this.
83const LOG_LIMIT: u64 = 5 * 1024 * 1024;
84/// The last lines of the log settings show.
85const 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)]
89pub 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)]
97struct 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)]
108struct 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)]
117pub 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")]
131pub 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.
154enum 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)]
167struct Registration {
168 server: String,
169 workspace: String,
170 backend: Backend,
171}
172
173fn 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
177impl 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.
194fn 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.
205fn 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
218fn 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
233pub 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.
257pub 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.
263fn 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
270fn 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.
277pub 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.
288pub fn settings_changed(app: &AppHandle) {
289 ask(app, Ask::Look);
290 publish(app);
291}
292
293/// Start or Stop, from settings or the tray.
294pub 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.
306pub 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.
312pub 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.
319pub 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.
335pub 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.
344pub 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.
379fn 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.
396pub 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.
401pub 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.
406pub 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.
417enum 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
424struct 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
442struct 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)]
456struct Extra {
457 detail: Option<String>,
458 exit_code: Option<i32>,
459 wait_seconds: Option<u64>,
460}
461
462fn detail(text: impl Into<String>) -> Extra {
463 Extra { detail: Some(text.into()), ..Extra::default() }
464}
465
466impl 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.
986fn 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)]
1004mod 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}