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

Merge pull request #183 from flagon-io/retry-lost-starts

A job whose sandbox never reports goes back in the queue instead of failing 10 minutes later

syntaqxcommitted Parentsbb14af6a6204c6Browse files
2 files+89−10/2 viewed
+34−0
8686 /// fails saying so. Tried again each minute until then.
8787 pub const WAIT_FOR_MACHINE_MS: u64 = 30 * 60 * 1000;
8888
89+/// How long a job on g1t's machines has to say anything after it starts.
90+/// Its runner reports its first step within seconds (6 to 12 on Deploy
91+/// #445); one that has said nothing by now never got a sandbox, so it is
92+/// tried again rather than left until SILENT_MS fails it. On 2026-10-20
93+/// two jobs of Deploy #443 were started from the request of a job that
94+/// reported its end: no container was free, the library's retries took
95+/// longer than that job's container stayed connected, and the start was
96+/// cancelled with the request, after the jobs were marked as running.
97+pub const FIRST_REPORT_MS: u64 = 3 * 60 * 1000;
98+
99+/// Whether a job that started at `started` and was last heard from at
100+/// `seen` (RFC 3339, both set as it starts) never reported, and is past
101+/// FIRST_REPORT_MS at `now`. Every report moves `seen` on.
102+pub fn never_reported(started: Option<&str>, seen: Option<&str>, now: u64) -> bool {
103+ match (started, seen) {
104+ (Some(started), Some(seen)) if started == seen => {
105+ g1t_contracts::time::parse_rfc3339(started).is_some_and(|started| now.saturating_sub(started) >= FIRST_REPORT_MS)
106+ }
107+ _ => false,
108+ }
109+}
110+
89111 /// Whether a job that has waited for a machine since `since` (RFC 3339)
90112 /// should stop waiting at `now`. One whose start cannot be read waits from
91113 /// now on.
463485 assert!(waited_for_a_machine(Some(&since), 1_000_000 + WAIT_FOR_MACHINE_MS));
464486 assert!(!waited_for_a_machine(Some("not a time"), u64::MAX));
465487 }
488+
489+ #[test]
490+ fn a_job_that_never_reported_is_lost_after_three_minutes() {
491+ let started = g1t_contracts::time::rfc3339(1_000_000);
492+ let later = g1t_contracts::time::rfc3339(1_005_000);
493+ assert!(!never_reported(Some(&started), Some(&started), 1_000_000 + FIRST_REPORT_MS - 1));
494+ assert!(never_reported(Some(&started), Some(&started), 1_000_000 + FIRST_REPORT_MS));
495+ // Once it has said anything, only SILENT_MS ends it.
496+ assert!(!never_reported(Some(&started), Some(&later), u64::MAX));
497+ assert!(!never_reported(None, None, u64::MAX));
498+ assert!(!never_reported(Some("not a time"), Some("not a time"), u64::MAX));
499+ }
466500 }
+55−1
15991599
16001600 /// Starts queued jobs, oldest first, while their workspace has room.
16011601 pub async fn start_queued(&self) -> Result<()> {
1602+ self.start_queued_except(&std::collections::HashSet::new()).await
1603+ }
1604+
1605+ /// [`Self::start_queued`], leaving `held` for a later sweep.
1606+ async fn start_queued_except(&self, held: &std::collections::HashSet<String>) -> Result<()> {
16021607 let queued = self
16031608 .db
16041609 // Self-hosted jobs are taken by their runners (runners.rs).
16191624 .map_or(0, |count| count.n)
16201625 };
16211626 for job in queued {
1627+ if held.contains(&job.id) {
1628+ continue;
1629+ }
16221630 let in_workspace = match running.get(&job.namespace) {
16231631 Some(n) => *n,
16241632 None => {
28672875 }
28682876 // Jobs whose sandbox went quiet or ran past their time.
28692877 let running = self.db.prepare("SELECT * FROM jobs WHERE status = 'in_progress'").all().await?.results::<JobRow>()?;
2878+ let mut lost = std::collections::HashSet::new();
28702879 for job in running {
28712880 // Times in g1t's format compare as text.
28722881 let before = |ms: u64| rfc3339(now_ms.saturating_sub(ms));
28852894 let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
28862895 }
28872896 self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2897+ } else if job.runner_id.is_none() && crate::never_reported(job.started_at.as_deref(), job.seen_at.as_deref(), now_ms) {
2898+ // Its sandbox never ran it: nothing it did needs undoing.
2899+ // It starts again at the next sweep, not this one: the same
2900+ // sandbox takes it, and the stop of the lost one finishes
2901+ // in that sandbox's alarm, after this.
2902+ if self.retry_lost_start(&job, now_ms).await? {
2903+ lost.insert(job.id.clone());
2904+ }
28882905 } else if silent {
28892906 let reason = match &job.runner_name {
28902907 Some(name) => format!("The self-hosted runner {name} stopped answering."),
29422959 {
29432960 worker::console_error!("actions: the logs' sweep failed: {error}");
29442961 }
2945− self.start_queued().await
2962+ self.start_queued_except(&lost).await
2963+ }
2964+
2965+ /// A job whose sandbox never reported (`never_reported`): back in the
2966+ /// queue, its token void so that sandbox can no longer report for it,
2967+ /// and the sandbox stopped, which gives back what billing held for it.
2968+ /// It waits as a job with no machine does, and fails once it has waited
2969+ /// WAIT_FOR_MACHINE_MS. Returns whether it went back in the queue.
2970+ async fn retry_lost_start(&self, job: &JobRow, now_ms: u64) -> Result<bool> {
2971+ let since = job.waiting_since.clone().or_else(|| job.started_at.clone()).unwrap_or_else(|| rfc3339(now_ms));
2972+ if crate::waited_for_a_machine(Some(&since), now_ms) {
2973+ let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2974+ let minutes = crate::WAIT_FOR_MACHINE_MS / 60_000;
2975+ let reason = format!("No machine ran the job in {minutes} minutes: each one it was given stopped answering before it started.");
2976+ self.finish_job(&job.id, "failure", Some(&reason), None).await?;
2977+ return Ok(false);
2978+ }
2979+ // Not if it reported after all, since it was read.
2980+ let requeued = self
2981+ .db
2982+ .prepare(
2983+ "UPDATE jobs SET status = 'queued', token_hash = NULL, first_token_hash = NULL, started_at = NULL, seen_at = NULL, waiting_since = ?, reason = ?
2984+ WHERE id = ? AND status = 'in_progress' AND seen_at = started_at RETURNING id",
2985+ )
2986+ .bind(&[
2987+ since.as_str().into(),
2988+ "Its machine stopped answering before it started; it waits for another.".into(),
2989+ job.id.as_str().into(),
2990+ ])?
2991+ .first::<Value>(None)
2992+ .await?;
2993+ if requeued.is_none() {
2994+ return Ok(false);
2995+ }
2996+ worker::console_warn!("actions: job {} never reported; it goes back in the queue", job.id);
2997+ self.revoke_job_tokens(&job.id).await;
2998+ let _: Result<Value> = g1t_kit::call(&self.runner, "stop_actions_job", &json!({ "job": job.id })).await;
2999+ Ok(true)
29463000 }
29473001 }
29483002

Comments

Sign in to comment.