Skip to content
Sign up

flagon-io/g1t

public
0
483 linesCodeBlameRaw
1//! Telling g1t how a job is going: its steps, its log in batches, its
2//! annotations, and how it ended. Every report carries the job's token.
3
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
6
7use anyhow::Result;
8use g1t_actions::mask;
9use serde_json::{Value, json};
10
11/// How long log lines wait before they are sent: about a second, so a
12/// running step's page follows it closely, at one write to g1t a second.
13const FLUSH_EVERY: Duration = Duration::from_millis(1000);
14/// How much log is sent at once.
15const FLUSH_BYTES: usize = 64 * 1024;
16/// The most one step may add to the job's summary, as on GitHub.
17pub(crate) const MAX_SUMMARY_BYTES: usize = 1024 * 1024;
18/// How long the runner goes without a report before it asks whether the
19/// job was cancelled.
20const PING_EVERY: Duration = Duration::from_secs(10);
21
22/// Whether the job was cancelled, as g1t's answers to its reports say.
23struct Cancel {
24 /// g1t said so.
25 said: AtomicBool,
26 /// The job has taken it in: the step it was on was stopped, and the
27 /// cleanup steps that follow are not stopped for it again.
28 taken: AtomicBool,
29}
30
31impl Cancel {
32 const fn new() -> Cancel {
33 Cancel { said: AtomicBool::new(false), taken: AtomicBool::new(false) }
34 }
35
36 /// Reads an answer to a report: `cancelled` when the run was cancelled.
37 fn hear(&self, answer: &Value) {
38 if answer["cancelled"].as_bool() == Some(true) {
39 self.said.store(true, Ordering::Relaxed);
40 }
41 }
42
43 fn cancelled(&self) -> bool {
44 self.said.load(Ordering::Relaxed)
45 }
46
47 fn interrupt(&self) -> bool {
48 self.cancelled() && !self.taken.load(Ordering::Relaxed)
49 }
50
51 fn take(&self) {
52 self.taken.store(true, Ordering::Relaxed);
53 }
54}
55
56/// This job's (one runs per process).
57static CANCEL: Cancel = Cancel::new();
58
59/// Whether g1t said the job was cancelled.
60pub(crate) fn cancelled() -> bool {
61 CANCEL.cancelled()
62}
63
64/// Whether a running step should be stopped: cancelled, and not yet taken in.
65pub(crate) fn interrupt() -> bool {
66 CANCEL.interrupt()
67}
68
69/// The job has seen the cancellation: what runs from here is its cleanup.
70pub(crate) fn take_cancel() {
71 CANCEL.take();
72}
73
74pub(crate) struct Api {
75 pub(crate) base: String,
76 pub(crate) job: String,
77 pub(crate) token: String,
78}
79
80impl Api {
81 pub(crate) fn spec(&self) -> Result<Value> {
82 let response = ureq::post(&format!("{}/actions/jobs/{}/spec", self.base, self.job))
83 .timeout(Duration::from_secs(60))
84 .send_json(json!({ "token": self.token }))?;
85 Ok(response.into_json()?)
86 }
87
88 /// Where to fetch another repository's action from: `{"source": "g1t",
89 /// "url", "ref", "token"}`, or `{"source": "github", "sha", "archive",
90 /// "note"}`: the commit GitHub's ref names, and whether g1t keeps its
91 /// archive (`action_archive`). Err holds why g1t
92 /// refused (`true`: the repository is private and may not be used
93 /// here), or that it could not be asked (`false`).
94 pub(crate) fn action(&self, repository: &str, git_ref: &str) -> std::result::Result<Value, (bool, String)> {
95 // The first job to use an action at a commit waits while g1t
96 // fetches and keeps its archive.
97 let sent = ureq::post(&format!("{}/actions/jobs/{}/action", self.base, self.job))
98 .timeout(Duration::from_secs(120))
99 .send_json(json!({ "token": self.token, "report": { "repository": repository, "ref": git_ref } }));
100 match sent {
101 Ok(response) => response.into_json().map_err(|error| (false, error.to_string())),
102 Err(ureq::Error::Status(code, response)) => {
103 let body: Value = response.into_json().unwrap_or(Value::Null);
104 let message = body["error"]["message"].as_str().unwrap_or("g1t did not answer.").to_owned();
105 Err((code == 403, message))
106 }
107 Err(error) => Err((false, error.to_string())),
108 }
109 }
110
111 /// Downloads the archive g1t keeps of `repository` at `sha` to `to`.
112 pub(crate) fn action_archive(&self, repository: &str, sha: &str, to: &std::path::Path) -> std::result::Result<u64, String> {
113 let response = ureq::post(&format!("{}/actions/jobs/{}/archive", self.base, self.job))
114 .timeout(Duration::from_secs(300))
115 .send_json(json!({ "token": self.token, "report": { "repository": repository, "sha": sha } }))
116 .map_err(|error| match error {
117 ureq::Error::Status(code, response) => {
118 let body: Value = response.into_json().unwrap_or(Value::Null);
119 body["error"]["message"].as_str().map_or_else(|| format!("g1t answered {code}"), str::to_owned)
120 }
121 other => other.to_string(),
122 })?;
123 let mut file = std::fs::File::create(to).map_err(|error| error.to_string())?;
124 std::io::copy(&mut response.into_reader(), &mut file).map_err(|error| error.to_string())
125 }
126
127 /// An API with no address, which sends nothing: what the tests use, so
128 /// a step's run is not held up by reports that go nowhere (each one
129 /// tried three times, with waits between, over a second and a half).
130 #[cfg(test)]
131 pub(crate) fn offline() -> Api {
132 Api { base: String::new(), job: "job_1".into(), token: "t".into() }
133 }
134
135 pub(crate) fn report(&self, report: Value) {
136 // No address: nowhere to send it (Api::offline).
137 if self.base.is_empty() {
138 return;
139 }
140 // A report that cannot be sent is tried a few times, then dropped:
141 // the job goes on, and g1t notices a silent job by itself.
142 for attempt in 0..3 {
143 let sent = ureq::post(&format!("{}/actions/jobs/{}", self.base, self.job))
144 .timeout(Duration::from_secs(30))
145 .send_json(json!({ "token": self.token, "report": report }));
146 match sent {
147 Ok(response) => {
148 if let Ok(answer) = response.into_json::<Value>() {
149 CANCEL.hear(&answer);
150 }
151 return;
152 }
153 // Refused: the job was cancelled or finished; nothing to retry.
154 Err(ureq::Error::Status(code, _)) if (400..500).contains(&code) => return,
155 Err(_) => std::thread::sleep(Duration::from_millis(500 * (attempt + 1))),
156 }
157 }
158 }
159}
160
161/// The job's log, masked, sent in batches. `masks` holds every form of
162/// every secret (`g1t_actions::mask`), longest first.
163pub(crate) struct Log {
164 pub(crate) api: Api,
165 pub(crate) masks: Vec<String>,
166 step: u32,
167 buffer: String,
168 last: Instant,
169 /// When anything was last sent, for the ping.
170 sent: Instant,
171 /// The problem matchers steps added, which last the whole job.
172 pub(crate) matchers: super::matchers::Matchers,
173 /// Everything sent so far, so a test reads the whole log however its
174 /// batches fell (`buffered`).
175 #[cfg(test)]
176 flushed: String,
177}
178
179impl Log {
180 pub(crate) fn new(api: Api, mut masks: Vec<String>) -> Log {
181 masks.retain(|mask| !mask.is_empty());
182 masks.sort_by_key(|mask| std::cmp::Reverse(mask.len()));
183 Log {
184 api,
185 masks,
186 step: 0,
187 buffer: String::new(),
188 last: Instant::now(),
189 sent: Instant::now(),
190 matchers: super::matchers::Matchers::default(),
191 #[cfg(test)]
192 flushed: String::new(),
193 }
194 }
195
196 /// The log so far, what was sent and what waits to be, without the
197 /// time each line is stamped with.
198 #[cfg(test)]
199 pub(crate) fn buffered(&self) -> String {
200 unstamped(&format!("{}{}", self.flushed, self.buffer))
201 }
202
203 /// Starts writing to step `number` (0 for the job's setup).
204 pub(crate) fn step(&mut self, number: u32) {
205 self.flush();
206 self.step = number;
207 }
208
209 pub(crate) fn mask(&self, text: &str) -> String {
210 mask::apply(text, &self.masks)
211 }
212
213 /// `::add-mask::`: masks `value` from here on, in every form it can
214 /// take (each line, base64, JSON-escaped), as a secret is.
215 pub(crate) fn add_mask(&mut self, value: &str) {
216 let mut added = false;
217 for variant in mask::variants(value) {
218 if !self.masks.contains(&variant) {
219 self.masks.push(variant);
220 added = true;
221 }
222 }
223 if added {
224 self.masks.sort_by_key(|mask| std::cmp::Reverse(mask.len()));
225 }
226 }
227
228 /// Adds a line to the log, masked, each of its lines stamped with the
229 /// time, as GitHub's runner writes its logs
230 /// (`2026-10-12T09:41:07.1234567Z text`).
231 pub(crate) fn line(&mut self, text: &str) {
232 let masked = self.mask(text);
233 let at = stamp(SystemTime::now());
234 for line in masked.split('\n') {
235 self.buffer.push_str(&at);
236 self.buffer.push(' ');
237 self.buffer.push_str(line);
238 self.buffer.push('\n');
239 }
240 if self.buffer.len() >= FLUSH_BYTES || self.last.elapsed() >= FLUSH_EVERY {
241 self.flush();
242 }
243 }
244
245 /// Sends what is waiting if it has waited long enough, and asks after
246 /// the job when nothing has been sent for a while, so a cancellation
247 /// reaches a step that prints nothing.
248 pub(crate) fn tick(&mut self) {
249 if !self.buffer.is_empty() && self.last.elapsed() >= FLUSH_EVERY {
250 self.flush();
251 } else if self.sent.elapsed() >= PING_EVERY {
252 self.sent = Instant::now();
253 self.api.report(json!({ "kind": "ping" }));
254 }
255 }
256
257 pub(crate) fn flush(&mut self) {
258 self.last = Instant::now();
259 if self.buffer.is_empty() {
260 return;
261 }
262 self.sent = Instant::now();
263 let text = std::mem::take(&mut self.buffer);
264 #[cfg(test)]
265 self.flushed.push_str(&text);
266 self.api.report(json!({ "kind": "log", "step": self.step, "text": text }));
267 }
268
269 pub(crate) fn steps(&self, names: &[String]) {
270 self.api.report(json!({ "kind": "steps", "steps": names }));
271 }
272
273 pub(crate) fn step_state(&mut self, number: u32, name: &str, status: &str, conclusion: Option<&str>) {
274 self.flush();
275 let name = self.mask(name);
276 self.api.report(json!({ "kind": "step", "number": number, "name": name, "status": status, "conclusion": conclusion }));
277 }
278
279 /// What a step wrote to `$GITHUB_STEP_SUMMARY`, masked, for the run's
280 /// page. More than 1 MiB is refused, as on GitHub, with an error in the
281 /// log.
282 pub(crate) fn summary(&mut self, markdown: &str) {
283 if markdown.trim().is_empty() {
284 return;
285 }
286 if markdown.len() > MAX_SUMMARY_BYTES {
287 self.line(&format!(
288 "##[error]$GITHUB_STEP_SUMMARY upload aborted: a step's summary may be up to 1024k, and this one is {}k.",
289 markdown.len().div_ceil(1024)
290 ));
291 return;
292 }
293 let markdown = self.mask(markdown);
294 self.flush();
295 self.api.report(json!({ "kind": "summary", "step": self.step, "markdown": markdown }));
296 }
297
298 /// An annotation, from `::error::` and its kin or a problem matcher:
299 /// its `file`, `line`, `col`, `endLine` and `endColumn` as GitHub's
300 /// commands spell them, sent as `file`, `line`, `column`, `end_line`
301 /// and `end_column`.
302 pub(crate) fn annotation(&mut self, level: &str, message: &str, properties: &serde_json::Map<String, Value>) {
303 let report = self.annotation_report(level, message, properties);
304 self.api.report(report);
305 }
306
307 fn annotation_report(&self, level: &str, message: &str, properties: &serde_json::Map<String, Value>) -> Value {
308 let message = self.mask(message);
309 let title = properties.get("title").and_then(Value::as_str).map(|title| self.mask(title));
310 let number = |keys: &[&str]| keys.iter().find_map(|key| properties.get(*key).and_then(Value::as_str)).and_then(|n| n.trim().parse::<u32>().ok());
311 json!({
312 "kind": "annotation",
313 "level": level,
314 "message": message,
315 "title": title,
316 "file": properties.get("file"),
317 "line": number(&["line"]),
318 "column": number(&["col"]),
319 "end_line": number(&["endLine"]),
320 "end_column": number(&["endColumn"]),
321 })
322 }
323
324 pub(crate) fn done(&mut self, conclusion: &str, outputs: &serde_json::Map<String, Value>, reason: Option<&str>) {
325 let outputs = self.withhold_secrets(outputs);
326 self.flush();
327 self.api.report(json!({ "kind": "done", "conclusion": conclusion, "outputs": outputs, "reason": reason }));
328 }
329
330 /// The job's outputs without any that hold a secret, as on GitHub: an
331 /// output goes to other jobs and to the run's page, where no mask
332 /// reaches. Each one left out is warned of in the log.
333 pub(crate) fn withhold_secrets(&mut self, outputs: &serde_json::Map<String, Value>) -> serde_json::Map<String, Value> {
334 let mut kept = serde_json::Map::new();
335 for (name, value) in outputs {
336 let text = match value {
337 Value::String(text) => text.clone(),
338 other => other.to_string(),
339 };
340 if mask::reveals(&text, &self.masks) {
341 self.line(&format!("##[warning]The output `{name}` was left out: it holds a secret."));
342 continue;
343 }
344 kept.insert(name.clone(), value.clone());
345 }
346 kept
347 }
348}
349
350/// A time as GitHub's logs stamp each line: RFC 3339 in UTC, to a tenth
351/// of a microsecond (`2026-10-12T09:41:07.1234567Z`).
352pub(crate) fn stamp(at: SystemTime) -> String {
353 let since = at.duration_since(UNIX_EPOCH).unwrap_or_default();
354 let seconds = since.as_secs();
355 let (days, rest) = (seconds / 86_400, seconds % 86_400);
356 let (year, month, day) = civil(days as i64);
357 format!(
358 "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}.{:07}Z",
359 rest / 3600,
360 rest % 3600 / 60,
361 rest % 60,
362 since.subsec_nanos() / 100
363 )
364}
365
366/// The calendar date `days` after 1970-01-01 (Howard Hinnant's
367/// `civil_from_days`).
368fn civil(days: i64) -> (i64, u32, u32) {
369 let z = days + 719_468;
370 let era = z.div_euclid(146_097);
371 let doe = z.rem_euclid(146_097);
372 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
373 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
374 let mp = (5 * doy + 2) / 153;
375 let day = (doy - (153 * mp + 2) / 5 + 1) as u32;
376 let month = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32;
377 let year = yoe + era * 400 + i64::from(month <= 2);
378 (year, month, day)
379}
380
381/// A log without the time stamped at the start of each line.
382#[cfg(test)]
383pub(crate) fn unstamped(text: &str) -> String {
384 text.split_inclusive('\n')
385 .map(|line| match line.split_once(' ') {
386 Some((at, rest)) if at.len() == 28 && at.ends_with('Z') && at.as_bytes()[10] == b'T' => rest,
387 _ => line,
388 })
389 .collect()
390}
391
392#[cfg(test)]
393mod tests {
394 use super::*;
395
396 #[test]
397 fn every_line_is_stamped_with_the_time_as_github_stamps_it() {
398 assert_eq!(stamp(UNIX_EPOCH), "1970-01-01T00:00:00.0000000Z");
399 assert_eq!(stamp(UNIX_EPOCH + Duration::new(1_791_797_067, 123_456_789)), "2026-10-12T09:24:27.1234567Z");
400 // A leap day, and the last day of February in a century that has none.
401 assert_eq!(stamp(UNIX_EPOCH + Duration::from_secs(951_782_400)), "2000-02-29T00:00:00.0000000Z");
402 assert_eq!(stamp(UNIX_EPOCH + Duration::from_secs(4_107_542_399)), "2100-02-28T23:59:59.0000000Z");
403 let mut log = log(&["s3cr3t"]);
404 log.line("one s3cr3t\ntwo");
405 let lines: Vec<&str> = log.buffer.lines().collect();
406 assert_eq!(lines.len(), 2, "each line of a message is stamped");
407 for (line, text) in lines.iter().zip(["one ***", "two"]) {
408 let (at, rest) = line.split_once(' ').unwrap();
409 assert_eq!((at.len(), &at[10..11], &at[27..]), (28, "T", "Z"), "{line}");
410 assert_eq!(rest, text);
411 }
412 assert_eq!(log.buffered(), "one ***\ntwo\n");
413 }
414
415 fn log(secrets: &[&str]) -> Log {
416 Log::new(Api::offline(), mask::all_variants(secrets.iter().copied()))
417 }
418
419 #[test]
420 fn outputs_holding_a_secret_are_left_out() {
421 let mut log = log(&["s3cr3t-token"]);
422 let mut outputs = serde_json::Map::new();
423 outputs.insert("version".into(), json!("1.2.0"));
424 outputs.insert("leak".into(), json!("token=s3cr3t-token"));
425 outputs.insert("encoded".into(), json!("czNjcjN0LXRva2Vu"));
426 let kept = log.withhold_secrets(&outputs);
427 assert_eq!(kept.keys().collect::<Vec<_>>(), ["version"]);
428 assert!(log.buffer.contains("The output `leak` was left out"));
429 assert!(log.buffer.contains("The output `encoded` was left out"));
430 }
431
432 #[test]
433 fn a_cancelled_answer_is_heard_once_and_taken_in() {
434 let cancel = Cancel::new();
435 cancel.hear(&json!({ "ok": true }));
436 assert!(!cancel.cancelled());
437 cancel.hear(&json!({ "ok": true, "cancelled": true }));
438 assert!(cancel.cancelled() && cancel.interrupt());
439 cancel.take();
440 assert!(cancel.cancelled() && !cancel.interrupt(), "cleanup steps are not stopped again");
441 cancel.hear(&json!({ "ok": true, "cancelled": false }));
442 assert!(cancel.cancelled(), "a cancelled job stays cancelled");
443 }
444
445 #[test]
446 fn a_summary_past_its_limit_is_refused_in_the_log() {
447 let mut log = log(&[]);
448 log.summary("
449");
450 assert!(log.buffer.is_empty(), "an empty summary is not sent");
451 log.summary(&"x".repeat(MAX_SUMMARY_BYTES + 1));
452 assert!(log.buffer.contains("$GITHUB_STEP_SUMMARY upload aborted"));
453 assert!(log.buffer.contains("1025k"));
454 }
455
456 #[test]
457 fn annotations_carry_their_columns_and_end() {
458 let log = log(&["hidden"]);
459 let properties: serde_json::Map<String, Value> =
460 serde_json::from_value(json!({ "file": "a.rs", "line": "3", "col": "7", "endLine": "4", "endColumn": "2", "title": "t hidden" })).unwrap();
461 let report = log.annotation_report("error", "bad hidden", &properties);
462 assert_eq!(report["line"], 3);
463 assert_eq!(report["column"], 7);
464 assert_eq!(report["end_line"], 4);
465 assert_eq!(report["end_column"], 2);
466 assert_eq!(report["title"], "t ***");
467 assert_eq!(report["message"], "bad ***");
468 let bare = log.annotation_report("notice", "x", &serde_json::Map::new());
469 assert!(bare["column"].is_null() && bare["file"].is_null());
470 }
471
472 #[test]
473 fn added_masks_cover_every_form() {
474 let mut log = log(&[]);
475 log.add_mask("line one\nline two");
476 assert_eq!(log.mask("first: line one"), "first: ***");
477 assert_eq!(log.mask("then line two"), "then ***");
478 // Longest first: the whole value, not its pieces.
479 log.add_mask("abc");
480 log.add_mask("abcdef");
481 assert_eq!(log.mask("abcdef"), "***");
482 }
483}