Skip to content
Sign up

flagon-io/g1t

public
00

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

Commit

A job's spec cut short is asked for again instead of failing the job

Deploy #449's core (rust-3) job failed in 5 seconds with "The runner could not fetch the job: Failed to read JSON: response body closed before all bytes were read": the runner asked for its spec once and gave the whole job up on one truncated answer. Api::spec now asks again, up to 4 times with 1, 2 and 4 seconds between, when the connection drops, the body ends early or g1t answers 5xx; never on a 4xx, which says no. Not with rotate (self-hosted runners): g1t spends the token on the first ask whether or not the answer arrived, so a second would be refused, as plan.rs's a_job_token_fetches_its_spec_once means it to be. g1t's own sandboxes take the runner from the image Deploy builds, so this reaches them with the next deploy.

Checked:
cargo test -p g1t-runner (262, new: a spec cut short is asked for again and arrives; one that spends its token is asked for once; only what may arrive next time is retried); cargo clippy -p g1t-runner --all-targets.
G1t-Session:
ses_01m5bgsckce7gt943v64h219yy
syntaqxcommitted Parent40a3065Browse files
1 file+129−20/1 viewed
+129−2
7171 CANCEL.take();
7272 }
7373
74+/// How many times a spec is asked for before the job fails.
75+const SPEC_TRIES: u32 = 4;
76+
77+/// How long to wait before asking for a spec again, after `attempt` tries:
78+/// 1, 2, then 4 seconds.
79+fn spec_backoff(attempt: u32) -> Duration {
80+ Duration::from_secs(1 << (attempt - 1).min(2))
81+}
82+
83+/// Why a spec did not arrive.
84+#[derive(Debug)]
85+enum SpecError {
86+ /// g1t answered with this status.
87+ Status(u16),
88+ /// The request could not be made or answered (connection, timeout).
89+ Transport(String),
90+ /// The answer's body did not read as JSON: cut short, usually.
91+ Body(std::io::Error),
92+}
93+
94+impl SpecError {
95+ /// Whether asking again may get the spec: not when g1t said no.
96+ fn worth_retrying(&self) -> bool {
97+ match self {
98+ SpecError::Status(code) => *code >= 500,
99+ SpecError::Transport(_) | SpecError::Body(_) => true,
100+ }
101+ }
102+}
103+
104+impl From<ureq::Error> for SpecError {
105+ fn from(error: ureq::Error) -> Self {
106+ match error {
107+ ureq::Error::Status(code, _) => SpecError::Status(code),
108+ ureq::Error::Transport(transport) => SpecError::Transport(transport.to_string()),
109+ }
110+ }
111+}
112+
113+impl std::fmt::Display for SpecError {
114+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
115+ match self {
116+ SpecError::Status(code) => write!(f, "g1t answered {code}"),
117+ SpecError::Transport(error) => write!(f, "{error}"),
118+ SpecError::Body(error) => write!(f, "Failed to read JSON: {error}"),
119+ }
120+ }
121+}
122+
123+impl std::error::Error for SpecError {}
124+
74125 #[derive(Clone)]
75126 pub(crate) struct Api {
76127 pub(crate) base: String,
88139 /// With `broker`, this harness passes its job's requests to brokered
89140 /// secrets' hosts through g1t (broker.rs), and the spec lists each
90141 /// stand-in's hosts (`brokered`).
142+ ///
143+ /// A spec that does not arrive whole (the connection drops, or the body
144+ /// ends early) or that g1t could not make (5xx) is asked for again, up
145+ /// to [`SPEC_TRIES`] times: Deploy #449's job failed on one truncated
146+ /// answer. Not with `rotate`: g1t spends the token on the first ask,
147+ /// whether or not the answer arrived, so a second would be refused.
91148 pub(crate) fn spec(&mut self, rotate: bool, broker: bool) -> Result<Value> {
92149 let mut body = json!({ "token": self.token });
93150 if rotate {
96153 if broker {
97154 body["broker"] = json!(true);
98155 }
99− let response = ureq::post(&format!("{}/actions/jobs/{}/spec", self.base, self.job)).timeout(Duration::from_secs(60)).send_json(body)?;
100− let mut spec: Value = response.into_json()?;
156+ let url = format!("{}/actions/jobs/{}/spec", self.base, self.job);
157+ let tries = if rotate { 1 } else { SPEC_TRIES };
158+ let mut attempt = 1;
159+ let mut spec: Value = loop {
160+ let fetched = ureq::post(&url)
161+ .timeout(Duration::from_secs(60))
162+ .send_json(body.clone())
163+ .map_err(SpecError::from)
164+ .and_then(|response| response.into_json::<Value>().map_err(SpecError::Body));
165+ match fetched {
166+ Ok(spec) => break spec,
167+ Err(error) if attempt < tries && error.worth_retrying() => {
168+ eprintln!("g1t-runner: the job's spec did not arrive whole ({error}); asking again");
169+ std::thread::sleep(spec_backoff(attempt));
170+ attempt += 1;
171+ }
172+ Err(error) => return Err(error.into()),
173+ }
174+ };
101175 self.take_next_token(&mut spec);
102176 Ok(spec)
103177 }
469543 mod tests {
470544 use super::*;
471545
546+ /// A g1t on 127.0.0.1 that answers each spec request with the next of
547+ /// `answers` (a body, and whether to cut it short), counting the asks.
548+ fn serve(answers: Vec<(&'static str, bool)>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
549+ use std::io::{Read, Write};
550+ let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
551+ let base = format!("http://{}", listener.local_addr().unwrap());
552+ let asked = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
553+ let count = asked.clone();
554+ std::thread::spawn(move || {
555+ for (body, cut) in answers {
556+ let Ok((mut stream, _)) = listener.accept() else { return };
557+ let mut request = [0u8; 4096];
558+ let _ = stream.read(&mut request);
559+ count.fetch_add(1, Ordering::SeqCst);
560+ // Cut short: says the whole length, sends half, and hangs up.
561+ let sent = if cut { &body[..body.len() / 2] } else { body };
562+ let head = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
563+ let _ = stream.write_all(head.as_bytes());
564+ let _ = stream.write_all(sent.as_bytes());
565+ }
566+ });
567+ (base, asked)
568+ }
569+
570+ #[test]
571+ fn a_spec_cut_short_is_asked_for_again() {
572+ let body = r#"{"job":"job_1","masks":[],"steps":[{"run":"cargo test"}]}"#;
573+ let (base, asked) = serve(vec![(body, true), (body, false)]);
574+ let mut api = Api { base, job: "job_1".into(), token: "t".into() };
575+ let spec = api.spec(false, false).unwrap();
576+ assert_eq!(spec["steps"][0]["run"], "cargo test");
577+ assert_eq!(asked.load(Ordering::SeqCst), 2);
578+ }
579+
580+ #[test]
581+ fn a_spec_that_spends_its_token_is_asked_for_once() {
582+ let body = r#"{"job":"job_1","masks":[],"next_token":"fresh"}"#;
583+ let (base, asked) = serve(vec![(body, true), (body, false)]);
584+ let mut api = Api { base, job: "job_1".into(), token: "t".into() };
585+ let error = api.spec(true, false).unwrap_err().to_string();
586+ assert!(error.starts_with("Failed to read JSON"), "{error}");
587+ assert_eq!(asked.load(Ordering::SeqCst), 1);
588+ }
589+
590+ #[test]
591+ fn only_what_may_arrive_next_time_is_asked_again() {
592+ assert!(SpecError::Status(502).worth_retrying());
593+ assert!(!SpecError::Status(403).worth_retrying());
594+ assert!(!SpecError::Status(404).worth_retrying());
595+ assert!(SpecError::Transport("connection reset".into()).worth_retrying());
596+ assert_eq!((spec_backoff(1), spec_backoff(2), spec_backoff(3)), (Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(4)));
597+ }
598+
472599 #[test]
473600 fn the_token_a_spec_carries_replaces_the_one_it_was_fetched_with() {
474601 let mut api = Api::offline();

Comments

Sign in to comment.