Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/crates/runner/src/actions/process.rs

211 lines7,911 bytesCodeBlame
1//! 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() {
76 log.masks.push(data.trim().to_owned());
77 }
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,
122}
123
124/// Runs the command, sending its output (stdout and stderr together, a
125/// line at a time) through `commands` to the log, until it ends or
126/// `timeout` passes.
127pub(crate) fn run(mut command: Command, timeout: Duration, log: &mut Log, commands: &mut Commands) -> std::io::Result<Ended> {
128 command.stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped());
129 let mut child = command.spawn()?;
130 let (sender, lines) = mpsc::channel::<String>();
131 let mut readers = Vec::new();
132 let pipes: Vec<Box<dyn Read + Send>> = vec![
133 Box::new(child.stdout.take().expect("piped")),
134 Box::new(child.stderr.take().expect("piped")),
135 ];
136 for pipe in pipes {
137 let sender = sender.clone();
138 readers.push(std::thread::spawn(move || {
139 let mut reader = BufReader::new(pipe);
140 let mut buffer = Vec::new();
141 loop {
142 buffer.clear();
143 match reader.read_until(b'\n', &mut buffer) {
144 Ok(0) | Err(_) => break,
145 Ok(_) => {
146 let text = String::from_utf8_lossy(&buffer);
147 let text = text.trim_end_matches(['\n', '\r']);
148 // A progress bar redraws with \r; keep its last state.
149 let text = text.rsplit('\r').next().unwrap_or(text);
150 if sender.send(text.to_owned()).is_err() {
151 break;
152 }
153 }
154 }
155 }
156 }));
157 }
158 drop(sender);
159 let deadline = Instant::now() + timeout;
160 let mut timed_out = false;
161 loop {
162 match lines.recv_timeout(Duration::from_millis(250)) {
163 Ok(line) => {
164 if let Some(shown) = commands.handle(&line, log) {
165 log.line(&shown);
166 }
167 }
168 Err(mpsc::RecvTimeoutError::Timeout) => log.tick(),
169 Err(mpsc::RecvTimeoutError::Disconnected) => break,
170 }
171 if Instant::now() >= deadline {
172 timed_out = true;
173 let _ = child.kill();
174 break;
175 }
176 }
177 let status = child.wait()?;
178 for reader in readers {
179 let _ = reader.join();
180 }
181 // Whatever arrived after the readers finished.
182 while let Ok(line) = lines.try_recv() {
183 if let Some(shown) = commands.handle(&line, log) {
184 log.line(&shown);
185 }
186 }
187 if timed_out {
188 return Ok(Ended::TimedOut);
189 }
190 Ok(Ended::Exited(status.code().unwrap_or(1)))
191}
192
193#[cfg(test)]
194mod tests {
195 use super::parse_command;
196
197 #[test]
198 fn commands_are_read_with_their_properties() {
199 let (name, properties, data) = parse_command("::error file=app.js,line=10,title=Bad%3A thing::Something%0Abroke").unwrap();
200 assert_eq!(name, "error");
201 assert_eq!(properties["file"], "app.js");
202 assert_eq!(properties["line"], "10");
203 assert_eq!(properties["title"], "Bad: thing");
204 assert_eq!(data, "Something\nbroke");
205 let (name, properties, data) = parse_command("::group::Install").unwrap();
206 assert_eq!((name.as_str(), properties.len(), data.as_str()), ("group", 0, "Install"));
207 assert_eq!(parse_command("::set-output name=version::1.2.3").unwrap().1["name"], "version");
208 assert!(parse_command("plain text").is_none());
209 assert!(parse_command(":: not a command").is_none());
210 }
211}