Skip to content
Request access

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 #303 from flagon-io/292-machine-retry

A job refused for want of a machine is tried again within seconds

syntaqxcommitted Parentsad2cfd28580872Browse files
2 files+42−90/2 viewed
+17−2
8383
8484 /// How long a job waits for a machine when the runner has none for it
8585 /// (every sandbox of its class taken, or a rollout replacing them) before it
86−/// fails saying so. Tried again each minute until then.
86+/// fails saying so. Tried again within seconds (MACHINE_RETRY_MS), then each
87+/// minute until then.
8788 pub const WAIT_FOR_MACHINE_MS: u64 = 30 * 60 * 1000;
8889
90+/// The pauses before a job refused for want of a machine is tried again,
91+/// after a request or the minute's sweep: refusals come in bursts that end
92+/// within seconds, and the next sweep left such a job waiting a median
93+/// 13.5 s and p90 120 s (#292). Together well under the 30 s a request's
94+/// `wait_until` may run after its answer.
95+pub const MACHINE_RETRY_MS: [u64; 3] = [2_000, 5_000, 10_000];
96+
8997 /// How long a job on g1t's machines has to say anything after it starts.
9098 /// Its runner reports its first step within seconds (6 to 12 on Deploy
9199 /// #445); one that has said nothing by now never got a sandbox, so it is
308316 ctx.wait_until(async move {
309317 match Actions::new(&env) {
310318 Ok(service) => {
311− if let Err(error) = service.start_queued().await {
319+ if let Err(error) = service.start_queued_retrying(&std::collections::HashSet::new()).await {
312320 worker::console_error!("actions: starting queued jobs after {method} failed: {error}");
313321 }
314322 }
519527 }
520528
521529 #[test]
530+ fn a_job_refused_a_machine_is_tried_again_within_seconds() {
531+ let waited: u64 = MACHINE_RETRY_MS.iter().sum();
532+ assert!(MACHINE_RETRY_MS.windows(2).all(|pair| pair[0] < pair[1]));
533+ assert!(waited < 30_000 && waited < WAIT_FOR_MACHINE_MS);
534+ }
535+
536+ #[test]
522537 fn a_job_that_never_reported_is_lost_after_three_minutes() {
523538 let started = g1t_contracts::time::rfc3339(1_000_000);
524539 let later = g1t_contracts::time::rfc3339(1_005_000);
+25−7
16041604 self.starts_wanted.set(true);
16051605 return Ok(());
16061606 }
1607− self.start_queued_except(&std::collections::HashSet::new()).await
1607+ self.start_queued_except(&std::collections::HashSet::new()).await.map(|_| ())
1608+ }
1609+
1610+ /// [`Self::start_queued`], and while the runner refuses jobs for want of
1611+ /// a machine, again after each of MACHINE_RETRY_MS rather than at the
1612+ /// next sweep (#292). Only where nothing waits on it: after a request's
1613+ /// answer, and at the end of the sweep.
1614+ pub async fn start_queued_retrying(&self, held: &std::collections::HashSet<String>) -> Result<()> {
1615+ for pause in crate::MACHINE_RETRY_MS {
1616+ if !self.start_queued_except(held).await? {
1617+ return Ok(());
1618+ }
1619+ worker::Delay::from(std::time::Duration::from_millis(pause)).await;
1620+ }
1621+ self.start_queued_except(held).await.map(|_| ())
16081622 }
16091623
1610− /// [`Self::start_queued`], leaving `held` for a later sweep.
1611− async fn start_queued_except(&self, held: &std::collections::HashSet<String>) -> Result<()> {
1624+ /// [`Self::start_queued`], leaving `held` for a later sweep. Returns
1625+ /// whether the runner refused any job for want of a machine.
1626+ async fn start_queued_except(&self, held: &std::collections::HashSet<String>) -> Result<bool> {
16121627 let queued = self
16131628 .db
16141629 // Self-hosted jobs are taken by their runners (runners.rs).
17271742 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")))
17281743 });
17291744 let started = futures_util::future::join_all(starts).await;
1745+ let mut no_machine = false;
17301746 for ((job, at, run, _), started) in claimed_jobs.into_iter().zip(started) {
17311747 if let Outcome::Fail(refused) = started {
17321748 // No machine free (every sandbox of its class taken, or a
1733− // rollout replacing them): back in the queue, tried again at
1734− // the next sweep, until it has waited WAIT_FOR_MACHINE_MS.
1749+ // rollout replacing them): back in the queue, tried again
1750+ // within seconds (start_queued_retrying) or at the next
1751+ // sweep, until it has waited WAIT_FOR_MACHINE_MS.
17351752 if refused.code == FailureCode::Unavailable {
17361753 let since = job.waiting_since.clone().unwrap_or_else(|| at.clone());
17371754 if crate::waited_for_a_machine(Some(&since), now_ms()) {
17491766 .run()
17501767 .await?;
17511768 worker::console_warn!("actions: job {} waits for a machine: {}", job.id, refused.message);
1769+ no_machine = true;
17521770 continue;
17531771 }
17541772 // Billing would not pay for a scheduled run's job: the
17671785 self.job_started(&job.id).await?;
17681786 }
17691787 }
1770− Ok(())
1788+ Ok(no_machine)
17711789 }
17721790
17731791 /// Finishes a job and moves its run along.
29772995 {
29782996 worker::console_error!("actions: the logs' sweep failed: {error}");
29792997 }
2980− self.start_queued_except(&lost).await
2998+ self.start_queued_retrying(&lost).await
29812999 }
29823000
29833001 /// A job whose sandbox never reported (`never_reported`): back in the

Comments

Sign in to comment.