Skip to content
Sign up

flagon-io/g1t

public
0
302 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)]
161 std::os::unix::process::CommandExt::process_group(&mut command, 0);
GitHub Actions on g1t, part two: running workflows162 let mut child = command.spawn()?;
163 let (sender, lines) = mpsc::channel::<String>();
164 let mut readers = Vec::new();
165 let pipes: Vec<Box<dyn Read + Send>> = vec![
166 Box::new(child.stdout.take().expect("piped")),
167 Box::new(child.stderr.take().expect("piped")),
168 ];
169 for pipe in pipes {
170 let sender = sender.clone();
171 readers.push(std::thread::spawn(move || {
172 let mut reader = BufReader::new(pipe);
173 let mut buffer = Vec::new();
174 loop {
175 buffer.clear();
176 match reader.read_until(b'\n', &mut buffer) {
177 Ok(0) | Err(_) => break,
178 Ok(_) => {
179 let text = String::from_utf8_lossy(&buffer);
180 let text = text.trim_end_matches(['\n', '\r']);
181 // A progress bar redraws with \r; keep its last state.
182 let text = text.rsplit('\r').next().unwrap_or(text);
183 if sender.send(text.to_owned()).is_err() {
184 break;
185 }
186 }
187 }
188 }
189 }));
190 }
191 drop(sender);
192 let deadline = Instant::now() + timeout;
193 let mut timed_out = false;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)194 // When the cancellation reached it, and which signal it has had.
195 let mut interrupted: Option<(Instant, u8)> = None;
GitHub Actions on g1t, part two: running workflows196 loop {
197 match lines.recv_timeout(Duration::from_millis(250)) {
198 Ok(line) => {
199 if let Some(shown) = commands.handle(&line, log) {
200 log.line(&shown);
201 }
202 }
Merge branch 'main' into actions-toolkit-oidc-artifacts203 Err(mpsc::RecvTimeoutError::Timeout) => {
204 // The job's Docker Engine starting, from its own thread.
205 for note in crate::docker::take_notes() {
206 log.line(&note);
207 }
208 log.tick();
209 }
GitHub Actions on g1t, part two: running workflows210 Err(mpsc::RecvTimeoutError::Disconnected) => break,
211 }
212 if Instant::now() >= deadline {
213 timed_out = true;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)214 signal(&child, "KILL");
GitHub Actions on g1t, part two: running workflows215 let _ = child.kill();
216 break;
217 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)218 // Cancelled: SIGINT, then SIGTERM, then killed, as on GitHub.
219 match interrupted {
220 None if interrupt() => {
221 log.line("##[error]The operation was canceled.");
222 signal(&child, "INT");
223 interrupted = Some((Instant::now(), 1));
224 }
225 Some((at, 1)) if at.elapsed() >= INTERRUPT_GRACE => {
226 signal(&child, "TERM");
227 interrupted = Some((Instant::now(), 2));
228 }
229 Some((at, 2)) if at.elapsed() >= TERMINATE_GRACE => {
230 signal(&child, "KILL");
231 let _ = child.kill();
232 break;
233 }
234 _ => {}
235 }
236 if interrupted.is_some() && matches!(child.try_wait(), Ok(Some(_))) {
237 // The step is gone; what it started may still hold its output.
238 signal(&child, "KILL");
239 break;
240 }
GitHub Actions on g1t, part two: running workflows241 }
242 let status = child.wait()?;
243 for reader in readers {
244 let _ = reader.join();
245 }
246 // Whatever arrived after the readers finished.
247 while let Ok(line) = lines.try_recv() {
248 if let Some(shown) = commands.handle(&line, log) {
249 log.line(&shown);
250 }
251 }
252 if timed_out {
253 return Ok(Ended::TimedOut);
254 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)255 if interrupted.is_some() {
256 return Ok(Ended::Cancelled);
257 }
GitHub Actions on g1t, part two: running workflows258 Ok(Ended::Exited(status.code().unwrap_or(1)))
259}
260
261#[cfg(test)]
262mod tests {
263 use super::parse_command;
264
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)265 /// A cancelled step hears SIGINT, and its own trap runs.
266 #[cfg(unix)]
267 #[test]
268 fn a_cancelled_step_is_interrupted_and_may_clean_up() {
269 use super::{Commands, Ended, run_until};
270 use crate::actions::report::{Api, Log};
271 use std::process::Command;
272 use std::time::{Duration, Instant};
273
274 let api = Api { base: "http://127.0.0.1:9".into(), job: "job_1".into(), token: "t".into() };
275 let mut log = Log::new(api, Vec::new());
276 let mut command = Command::new("sh");
277 command.args(["-c", "trap 'echo cleaned up; exit 3' INT; echo started; while true; do sleep 0.1; done"]);
278 let began = Instant::now();
279 let ask = move || began.elapsed() >= Duration::from_millis(600);
280 let ended = run_until(command, Duration::from_secs(60), &mut log, &mut Commands::default(), &ask).unwrap();
281 assert!(matches!(ended, Ended::Cancelled));
282 assert!(began.elapsed() < Duration::from_secs(8), "SIGINT ended it, not the kill after the grace period");
283 let text = log.buffered();
284 assert!(text.contains("The operation was canceled."), "{text}");
285 assert!(text.contains("cleaned up"), "{text}");
286 }
287
GitHub Actions on g1t, part two: running workflows288 #[test]
289 fn commands_are_read_with_their_properties() {
290 let (name, properties, data) = parse_command("::error file=app.js,line=10,title=Bad%3A thing::Something%0Abroke").unwrap();
291 assert_eq!(name, "error");
292 assert_eq!(properties["file"], "app.js");
293 assert_eq!(properties["line"], "10");
294 assert_eq!(properties["title"], "Bad: thing");
295 assert_eq!(data, "Something\nbroke");
296 let (name, properties, data) = parse_command("::group::Install").unwrap();
297 assert_eq!((name.as_str(), properties.len(), data.as_str()), ("group", 0, "Install"));
298 assert_eq!(parse_command("::set-output name=version::1.2.3").unwrap().1["name"], "version");
299 assert!(parse_command("plain text").is_none());
300 assert!(parse_command(":: not a command").is_none());
301 }
302}

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