Skip to content
Sign up

flagon-io/g1t

public
0
318 linesCodeBlameRaw

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.

GitHub Actions on g1t, part two: running workflows1//! Running one process for a step: its output streamed to the log as it
2//! comes, with GitHub's workflow commands (`::error::`, `::group::`,
3//! `::add-mask::`…) read out of it.
4
5use std::collections::BTreeMap;
6use std::io::{BufRead, BufReader, Read};
7use std::process::{Command, Stdio};
8use std::sync::mpsc;
9use std::time::{Duration, Instant};
10
11use serde_json::{Map, Value};
12
13use super::report::Log;
14
15/// What a step's workflow commands left behind.
16#[derive(Default)]
17pub(crate) struct Commands {
18 /// `::set-output` (old, still honoured).
19 pub(crate) outputs: BTreeMap<String, String>,
20 /// `::save-state`, for the action's post step.
21 pub(crate) state: BTreeMap<String, String>,
22 /// Set by `::stop-commands::token` until `::token::`.
23 pub(crate) stopped: Option<String>,
24 /// Whether `::debug::` lines are shown (`ACTIONS_STEP_DEBUG`).
25 pub(crate) debug: bool,
26}
27
28/// `%25`, `%0D`, `%0A`, and in properties `%3A` and `%2C`, as the
29/// toolkit escapes them.
30fn unescape(text: &str, property: bool) -> String {
31 let mut out = text.replace("%0D", "\r").replace("%0A", "\n");
32 if property {
33 out = out.replace("%3A", ":").replace("%2C", ",");
34 }
35 out.replace("%25", "%")
36}
37
38/// `::name key=value,key=value::message`, if the line is a command.
39pub(crate) fn parse_command(line: &str) -> Option<(String, Map<String, Value>, String)> {
40 let rest = line.trim_start().strip_prefix("::")?;
41 let end = rest.find("::")?;
42 let (head, data) = (&rest[..end], &rest[end + 2..]);
43 let (name, properties) = match head.split_once(' ') {
44 Some((name, properties)) => (name, properties),
45 None => (head, ""),
46 };
47 if name.is_empty() || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
48 return None;
49 }
50 let mut map = Map::new();
51 for pair in properties.split(',').filter(|p| !p.trim().is_empty()) {
52 if let Some((key, value)) = pair.split_once('=') {
53 map.insert(key.trim().to_owned(), Value::String(unescape(value, true)));
54 }
55 }
56 Some((name.to_owned(), map, unescape(data, false)))
57}
58
59impl Commands {
60 /// Handles one line of output: a command is acted on, and the line to
61 /// show (if any) is returned.
62 pub(crate) fn handle(&mut self, line: &str, log: &mut Log) -> Option<String> {
63 if let Some(token) = &self.stopped {
64 if line.trim() == format!("::{token}::") {
65 self.stopped = None;
66 return None;
67 }
68 return Some(line.to_owned());
69 }
70 let Some((name, properties, data)) = parse_command(line) else {
71 return Some(line.to_owned());
72 };
73 match name.as_str() {
74 "add-mask" => {
75 if !data.trim().is_empty() {
Merge branch 'worktree-agent-a3abfcce648e87dca'76 log.add_mask(data.trim());
GitHub Actions on g1t, part two: running workflows77 }
78 None
79 }
80 "error" | "warning" | "notice" => {
81 log.annotation(&name, &data, &properties);
82 let label = match name.as_str() {
83 "error" => "Error",
84 "warning" => "Warning",
85 _ => "Notice",
86 };
87 Some(format!("##[{name}]{label}: {data}"))
88 }
89 "group" => Some(format!("##[group]{data}")),
90 "endgroup" => Some("##[endgroup]".to_owned()),
91 "debug" => self.debug.then(|| format!("##[debug]{data}")),
92 "set-output" => {
93 if let Some(name) = properties.get("name").and_then(Value::as_str) {
94 self.outputs.insert(name.to_owned(), data);
95 }
96 None
97 }
98 "save-state" => {
99 if let Some(name) = properties.get("name").and_then(Value::as_str) {
100 self.state.insert(name.to_owned(), data);
101 }
102 None
103 }
104 "stop-commands" => {
105 self.stopped = Some(data);
106 None
107 }
108 "echo" => None,
109 "add-path" | "set-env" => Some(format!(
110 "##[error]The `{name}` command is disabled, as on GitHub. Write to the file in $GITHUB_{} instead.",
111 if name == "add-path" { "PATH" } else { "ENV" }
112 )),
113 _ => Some(line.to_owned()),
114 }
115 }
116}
117
118/// How a process ended.
119pub(crate) enum Ended {
120 Exited(i32),
121 TimedOut,
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)122 /// The run was cancelled: it was interrupted, then stopped.
123 Cancelled,
124}
125
126/// After a cancellation, how long a step has after SIGINT before SIGTERM,
127/// and after SIGTERM before it is killed, as GitHub's runner waits.
128const INTERRUPT_GRACE: Duration = Duration::from_millis(7500);
129const TERMINATE_GRACE: Duration = Duration::from_millis(2500);
130
131/// Sends `signal` to the process's group (it leads its own), so what the
132/// step started hears it too. Windows has no signals: it is left to `kill`.
133fn signal(child: &std::process::Child, signal: &str) {
134 if cfg!(unix) {
135 let _ = Command::new("kill")
136 .args([format!("-{signal}"), "--".into(), format!("-{}", child.id())])
137 .stdout(Stdio::null())
138 .stderr(Stdio::null())
139 .status();
140 }
GitHub Actions on g1t, part two: running workflows141}
142
143/// Runs the command, sending its output (stdout and stderr together, a
144/// line at a time) through `commands` to the log, until it ends or
145/// `timeout` passes.
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)146pub(crate) fn run(command: Command, timeout: Duration, log: &mut Log, commands: &mut Commands) -> std::io::Result<Ended> {
147 run_until(command, timeout, log, commands, &super::report::interrupt)
148}
149
150/// `run`, stopping the process gracefully once `interrupt` says so.
151fn run_until(
152 mut command: Command,
153 timeout: Duration,
154 log: &mut Log,
155 commands: &mut Commands,
156 interrupt: &dyn Fn() -> bool,
157) -> std::io::Result<Ended> {
GitHub Actions on g1t, part two: running workflows158 command.stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped());
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)159 // A group of its own, so a cancellation reaches what the step started.
160 #[cfg(unix)]
A cancelled step's cleanup trap runs even when the runner was started with SIGINT ignored, and the backup test tags with its own identity161 {
162 use std::os::unix::process::CommandExt;
163 command.process_group(0);
164 // A shell cannot trap a signal it was started with ignored, and a
165 // runner started in the background (as a service, or under `&`)
166 // passes SIGINT on ignored. Back to the defaults, so a cancelled
167 // step hears SIGINT and its own trap runs.
168 // SAFETY: only signal(), which is async-signal-safe, between fork and exec.
169 unsafe {
170 command.pre_exec(|| {
171 for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGQUIT] {
172 libc::signal(signal, libc::SIG_DFL);
173 }
174 Ok(())
175 });
176 }
177 }
GitHub Actions on g1t, part two: running workflows178 let mut child = command.spawn()?;
179 let (sender, lines) = mpsc::channel::<String>();
180 let mut readers = Vec::new();
181 let pipes: Vec<Box<dyn Read + Send>> = vec![
182 Box::new(child.stdout.take().expect("piped")),
183 Box::new(child.stderr.take().expect("piped")),
184 ];
185 for pipe in pipes {
186 let sender = sender.clone();
187 readers.push(std::thread::spawn(move || {
188 let mut reader = BufReader::new(pipe);
189 let mut buffer = Vec::new();
190 loop {
191 buffer.clear();
192 match reader.read_until(b'\n', &mut buffer) {
193 Ok(0) | Err(_) => break,
194 Ok(_) => {
195 let text = String::from_utf8_lossy(&buffer);
196 let text = text.trim_end_matches(['\n', '\r']);
197 // A progress bar redraws with \r; keep its last state.
198 let text = text.rsplit('\r').next().unwrap_or(text);
199 if sender.send(text.to_owned()).is_err() {
200 break;
201 }
202 }
203 }
204 }
205 }));
206 }
207 drop(sender);
208 let deadline = Instant::now() + timeout;
209 let mut timed_out = false;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)210 // When the cancellation reached it, and which signal it has had.
211 let mut interrupted: Option<(Instant, u8)> = None;
GitHub Actions on g1t, part two: running workflows212 loop {
213 match lines.recv_timeout(Duration::from_millis(250)) {
214 Ok(line) => {
215 if let Some(shown) = commands.handle(&line, log) {
216 log.line(&shown);
217 }
218 }
Merge branch 'main' into actions-toolkit-oidc-artifacts219 Err(mpsc::RecvTimeoutError::Timeout) => {
220 // The job's Docker Engine starting, from its own thread.
221 for note in crate::docker::take_notes() {
222 log.line(&note);
223 }
224 log.tick();
225 }
GitHub Actions on g1t, part two: running workflows226 Err(mpsc::RecvTimeoutError::Disconnected) => break,
227 }
228 if Instant::now() >= deadline {
229 timed_out = true;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)230 signal(&child, "KILL");
GitHub Actions on g1t, part two: running workflows231 let _ = child.kill();
232 break;
233 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)234 // Cancelled: SIGINT, then SIGTERM, then killed, as on GitHub.
235 match interrupted {
236 None if interrupt() => {
237 log.line("##[error]The operation was canceled.");
238 signal(&child, "INT");
239 interrupted = Some((Instant::now(), 1));
240 }
241 Some((at, 1)) if at.elapsed() >= INTERRUPT_GRACE => {
242 signal(&child, "TERM");
243 interrupted = Some((Instant::now(), 2));
244 }
245 Some((at, 2)) if at.elapsed() >= TERMINATE_GRACE => {
246 signal(&child, "KILL");
247 let _ = child.kill();
248 break;
249 }
250 _ => {}
251 }
252 if interrupted.is_some() && matches!(child.try_wait(), Ok(Some(_))) {
253 // The step is gone; what it started may still hold its output.
254 signal(&child, "KILL");
255 break;
256 }
GitHub Actions on g1t, part two: running workflows257 }
258 let status = child.wait()?;
259 for reader in readers {
260 let _ = reader.join();
261 }
262 // Whatever arrived after the readers finished.
263 while let Ok(line) = lines.try_recv() {
264 if let Some(shown) = commands.handle(&line, log) {
265 log.line(&shown);
266 }
267 }
268 if timed_out {
269 return Ok(Ended::TimedOut);
270 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)271 if interrupted.is_some() {
272 return Ok(Ended::Cancelled);
273 }
GitHub Actions on g1t, part two: running workflows274 Ok(Ended::Exited(status.code().unwrap_or(1)))
275}
276
277#[cfg(test)]
278mod tests {
279 use super::parse_command;
280
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)281 /// A cancelled step hears SIGINT, and its own trap runs.
282 #[cfg(unix)]
283 #[test]
284 fn a_cancelled_step_is_interrupted_and_may_clean_up() {
285 use super::{Commands, Ended, run_until};
286 use crate::actions::report::{Api, Log};
287 use std::process::Command;
288 use std::time::{Duration, Instant};
289
290 let api = Api { base: "http://127.0.0.1:9".into(), job: "job_1".into(), token: "t".into() };
291 let mut log = Log::new(api, Vec::new());
292 let mut command = Command::new("sh");
293 command.args(["-c", "trap 'echo cleaned up; exit 3' INT; echo started; while true; do sleep 0.1; done"]);
294 let began = Instant::now();
295 let ask = move || began.elapsed() >= Duration::from_millis(600);
296 let ended = run_until(command, Duration::from_secs(60), &mut log, &mut Commands::default(), &ask).unwrap();
297 assert!(matches!(ended, Ended::Cancelled));
298 assert!(began.elapsed() < Duration::from_secs(8), "SIGINT ended it, not the kill after the grace period");
299 let text = log.buffered();
300 assert!(text.contains("The operation was canceled."), "{text}");
301 assert!(text.contains("cleaned up"), "{text}");
302 }
303
GitHub Actions on g1t, part two: running workflows304 #[test]
305 fn commands_are_read_with_their_properties() {
306 let (name, properties, data) = parse_command("::error file=app.js,line=10,title=Bad%3A thing::Something%0Abroke").unwrap();
307 assert_eq!(name, "error");
308 assert_eq!(properties["file"], "app.js");
309 assert_eq!(properties["line"], "10");
310 assert_eq!(properties["title"], "Bad: thing");
311 assert_eq!(data, "Something\nbroke");
312 let (name, properties, data) = parse_command("::group::Install").unwrap();
313 assert_eq!((name.as_str(), properties.len(), data.as_str()), ("group", 0, "Install"));
314 assert_eq!(parse_command("::set-output name=version::1.2.3").unwrap().1["name"], "version");
315 assert!(parse_command("plain text").is_none());
316 assert!(parse_command(":: not a command").is_none());
317 }
318}

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