Skip to content
Sign up

flagon-io/g1t

public
0
333 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.

Pull requests say where they are live; the project menu shows what is coming1//! 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());
Pull requests say where they are live; the project menu shows what is coming77 }
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`.
Merge remote-tracking branch 'origin/main' into workspace-chat133/// Sent with kill(2) itself: a machine without the `kill` program (procps
134/// is not in every image) would otherwise give the step no SIGINT at all,
135/// and leave what it started running after the step was killed.
136#[cfg(unix)]
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)137fn signal(child: &std::process::Child, signal: &str) {
Merge remote-tracking branch 'origin/main' into workspace-chat138 let number = match signal {
139 "INT" => libc::SIGINT,
140 "TERM" => libc::SIGTERM,
141 _ => libc::SIGKILL,
142 };
143 let Ok(group) = libc::pid_t::try_from(child.id()) else {
144 return;
145 };
146 // SAFETY: kill(2) with a negative pid signals that process group; it
147 // reads no memory of ours.
148 unsafe {
149 libc::kill(-group, number);
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)150 }
Pull requests say where they are live; the project menu shows what is coming151}
152
Merge remote-tracking branch 'origin/main' into workspace-chat153#[cfg(not(unix))]
154fn signal(_child: &std::process::Child, _signal: &str) {}
155
Pull requests say where they are live; the project menu shows what is coming156/// Runs the command, sending its output (stdout and stderr together, a
157/// line at a time) through `commands` to the log, until it ends or
158/// `timeout` passes.
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)159pub(crate) fn run(command: Command, timeout: Duration, log: &mut Log, commands: &mut Commands) -> std::io::Result<Ended> {
160 run_until(command, timeout, log, commands, &super::report::interrupt)
161}
162
163/// `run`, stopping the process gracefully once `interrupt` says so.
164fn run_until(
165 mut command: Command,
166 timeout: Duration,
167 log: &mut Log,
168 commands: &mut Commands,
169 interrupt: &dyn Fn() -> bool,
170) -> std::io::Result<Ended> {
Pull requests say where they are live; the project menu shows what is coming171 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)172 // A group of its own, so a cancellation reaches what the step started.
173 #[cfg(unix)]
Merge remote-tracking branch 'origin/main' into workspace-chat174 {
175 use std::os::unix::process::CommandExt;
176 command.process_group(0);
177 // A shell cannot trap a signal it was started with ignored, and a
178 // runner started in the background (as a service, or under `&`)
179 // passes SIGINT on ignored. Back to the defaults, so a cancelled
180 // step hears SIGINT and its own trap runs.
181 // SAFETY: only signal(), which is async-signal-safe, between fork and exec.
182 unsafe {
183 command.pre_exec(|| {
184 for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGQUIT] {
185 libc::signal(signal, libc::SIG_DFL);
186 }
187 Ok(())
188 });
189 }
190 }
Pull requests say where they are live; the project menu shows what is coming191 let mut child = command.spawn()?;
192 let (sender, lines) = mpsc::channel::<String>();
193 let mut readers = Vec::new();
194 let pipes: Vec<Box<dyn Read + Send>> = vec![
195 Box::new(child.stdout.take().expect("piped")),
196 Box::new(child.stderr.take().expect("piped")),
197 ];
198 for pipe in pipes {
199 let sender = sender.clone();
200 readers.push(std::thread::spawn(move || {
201 let mut reader = BufReader::new(pipe);
202 let mut buffer = Vec::new();
203 loop {
204 buffer.clear();
205 match reader.read_until(b'\n', &mut buffer) {
206 Ok(0) | Err(_) => break,
207 Ok(_) => {
208 let text = String::from_utf8_lossy(&buffer);
209 let text = text.trim_end_matches(['\n', '\r']);
210 // A progress bar redraws with \r; keep its last state.
211 let text = text.rsplit('\r').next().unwrap_or(text);
212 if sender.send(text.to_owned()).is_err() {
213 break;
214 }
215 }
216 }
217 }
218 }));
219 }
220 drop(sender);
221 let deadline = Instant::now() + timeout;
222 let mut timed_out = false;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)223 // When the cancellation reached it, and which signal it has had.
224 let mut interrupted: Option<(Instant, u8)> = None;
Pull requests say where they are live; the project menu shows what is coming225 loop {
226 match lines.recv_timeout(Duration::from_millis(250)) {
227 Ok(line) => {
228 if let Some(shown) = commands.handle(&line, log) {
229 log.line(&shown);
230 }
231 }
Merge branch 'main' into actions-toolkit-oidc-artifacts232 Err(mpsc::RecvTimeoutError::Timeout) => {
233 // The job's Docker Engine starting, from its own thread.
234 for note in crate::docker::take_notes() {
235 log.line(&note);
236 }
237 log.tick();
238 }
Pull requests say where they are live; the project menu shows what is coming239 Err(mpsc::RecvTimeoutError::Disconnected) => break,
240 }
241 if Instant::now() >= deadline {
242 timed_out = true;
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)243 signal(&child, "KILL");
Pull requests say where they are live; the project menu shows what is coming244 let _ = child.kill();
245 break;
246 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)247 // Cancelled: SIGINT, then SIGTERM, then killed, as on GitHub.
248 match interrupted {
249 None if interrupt() => {
250 log.line("##[error]The operation was canceled.");
251 signal(&child, "INT");
252 interrupted = Some((Instant::now(), 1));
253 }
254 Some((at, 1)) if at.elapsed() >= INTERRUPT_GRACE => {
255 signal(&child, "TERM");
256 interrupted = Some((Instant::now(), 2));
257 }
258 Some((at, 2)) if at.elapsed() >= TERMINATE_GRACE => {
259 signal(&child, "KILL");
260 let _ = child.kill();
261 break;
262 }
263 _ => {}
264 }
265 if interrupted.is_some() && matches!(child.try_wait(), Ok(Some(_))) {
266 // The step is gone; what it started may still hold its output.
267 signal(&child, "KILL");
268 break;
269 }
Pull requests say where they are live; the project menu shows what is coming270 }
271 let status = child.wait()?;
272 for reader in readers {
273 let _ = reader.join();
274 }
275 // Whatever arrived after the readers finished.
276 while let Ok(line) = lines.try_recv() {
277 if let Some(shown) = commands.handle(&line, log) {
278 log.line(&shown);
279 }
280 }
281 if timed_out {
282 return Ok(Ended::TimedOut);
283 }
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)284 if interrupted.is_some() {
285 return Ok(Ended::Cancelled);
286 }
Pull requests say where they are live; the project menu shows what is coming287 Ok(Ended::Exited(status.code().unwrap_or(1)))
288}
289
290#[cfg(test)]
291mod tests {
292 use super::parse_command;
293
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)294 /// A cancelled step hears SIGINT, and its own trap runs.
295 #[cfg(unix)]
296 #[test]
297 fn a_cancelled_step_is_interrupted_and_may_clean_up() {
298 use super::{Commands, Ended, run_until};
299 use crate::actions::report::{Api, Log};
300 use std::process::Command;
301 use std::time::{Duration, Instant};
302
The runner's cancellation test no longer fails on a busy machine. It sent its log to an address that is not there (127.0.0.1:9), and once a step outlasted a batch of the log (1.5 s), each batch was tried three times with waits between, about 3 s, inside the loop that times the grace after SIGINT; two batches carried the step past the 8 s the test allows, and it read as SIGINT not ending the step. The tests now use an API with no address, which sends nothing, and the log keeps what it would have sent so the test still reads it all. Failing the same way on demand (cancelling at 1.6 s instead of 0.6 s) took 10.6 s before and passes in 1.8 s now.303 // Offline: on a busy machine the step can outlast a batch of the log,
304 // and sending it to an address that is not there held the loop for
305 // seconds, past the grace this test measures.
306 let mut log = Log::new(Api::offline(), Vec::new());
Merge Actions runs: summaries, attempts and re-runs, graceful cancel, log downloads, badges (actions 0009)307 let mut command = Command::new("sh");
308 command.args(["-c", "trap 'echo cleaned up; exit 3' INT; echo started; while true; do sleep 0.1; done"]);
309 let began = Instant::now();
310 let ask = move || began.elapsed() >= Duration::from_millis(600);
311 let ended = run_until(command, Duration::from_secs(60), &mut log, &mut Commands::default(), &ask).unwrap();
312 assert!(matches!(ended, Ended::Cancelled));
313 assert!(began.elapsed() < Duration::from_secs(8), "SIGINT ended it, not the kill after the grace period");
314 let text = log.buffered();
315 assert!(text.contains("The operation was canceled."), "{text}");
316 assert!(text.contains("cleaned up"), "{text}");
317 }
318
Pull requests say where they are live; the project menu shows what is coming319 #[test]
320 fn commands_are_read_with_their_properties() {
321 let (name, properties, data) = parse_command("::error file=app.js,line=10,title=Bad%3A thing::Something%0Abroke").unwrap();
322 assert_eq!(name, "error");
323 assert_eq!(properties["file"], "app.js");
324 assert_eq!(properties["line"], "10");
325 assert_eq!(properties["title"], "Bad: thing");
326 assert_eq!(data, "Something\nbroke");
327 let (name, properties, data) = parse_command("::group::Install").unwrap();
328 assert_eq!((name.as_str(), properties.len(), data.as_str()), ("group", 0, "Install"));
329 assert_eq!(parse_command("::set-output name=version::1.2.3").unwrap().1["name"], "version");
330 assert!(parse_command("plain text").is_none());
331 assert!(parse_command(":: not a command").is_none());
332 }
333}

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