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

A job Cloudflare has no container for waits for a machine instead of failing #181

Mergedsyntaqx merged wait-for-a-machine into main
7 files+83−40/7 viewed
+2−2
37533753
37543754 [[package]]
37553755 name = "serde_json"
3756−version = "1.0.154"
3756+version = "1.0.152"
37573757 source = "registry+https://github.com/rust-lang/crates.io-index"
3758−checksum = "e7e9cc8b1b85264074fbcc02a88680c4096b1e47df8f739dceb03bf482f04bd6"
3758+checksum = "1741ab7a6cc54a03a89b5d563ed60075c277d9e3cfa73ad0c1f23f23974703c6"
37593759 dependencies = [
37603760 "itoa",
37613761 "memchr",
+6−0
1+-- A job the runner had no machine for (Cloudflare gave its sandbox no
2+-- container: every instance taken, or a rollout replacing them) goes back
3+-- in the queue, and is tried again each minute. This says since when, so a
4+-- job that never gets one fails after WAIT_FOR_MACHINE_MS (src/lib.rs)
5+-- rather than waiting for ever.
6+ALTER TABLE jobs ADD COLUMN waiting_since TEXT;
+23−0
8080 in_workspace < RUNNING_PER_WORKSPACE
8181 || (in_workspace < BURST_PER_WORKSPACE && in_pool + POOL_HEADROOM < HOSTED_POOL)
8282 }
83+
84+/// How long a job waits for a machine when the runner has none for it
85+/// (every sandbox of its class taken, or a rollout replacing them) before it
86+/// fails saying so. Tried again each minute until then.
87+pub const WAIT_FOR_MACHINE_MS: u64 = 30 * 60 * 1000;
88+
89+/// Whether a job that has waited for a machine since `since` (RFC 3339)
90+/// should stop waiting at `now`. One whose start cannot be read waits from
91+/// now on.
92+pub fn waited_for_a_machine(since: Option<&str>, now: u64) -> bool {
93+ since
94+ .and_then(g1t_contracts::time::parse_rfc3339)
95+ .is_some_and(|since| now.saturating_sub(since) >= WAIT_FOR_MACHINE_MS)
96+}
8397 /// The longest a job on g1t's own machines may run, whatever its
8498 /// `timeout-minutes`.
8599 pub const MAX_TIMEOUT_MINUTES: u32 = g1t_actions::workflow::HOSTED_MAX_TIMEOUT_MINUTES;
440454 assert!(!has_room(RUNNING_PER_WORKSPACE, HOSTED_POOL - POOL_HEADROOM));
441455 assert!(!has_room(BURST_PER_WORKSPACE, 0));
442456 }
457+
458+ #[test]
459+ fn a_job_waits_for_a_machine_for_half_an_hour() {
460+ let since = g1t_contracts::time::rfc3339(1_000_000);
461+ assert!(!waited_for_a_machine(None, u64::MAX));
462+ assert!(!waited_for_a_machine(Some(&since), 1_000_000 + WAIT_FOR_MACHINE_MS - 1));
463+ assert!(waited_for_a_machine(Some(&since), 1_000_000 + WAIT_FOR_MACHINE_MS));
464+ assert!(!waited_for_a_machine(Some("not a time"), u64::MAX));
465+ }
443466 }
+29−1
253253 /// for another (migration 0020): good for `brokered_secrets` only.
254254 #[serde(default)]
255255 pub first_token_hash: Option<String>,
256+ /// Since when it has waited for a machine the runner had none of
257+ /// (migration 0025).
258+ #[serde(default)]
259+ pub waiting_since: Option<String>,
256260 }
257261
258262 /// How long a cancelled job has to run its `if: always()` and `cancelled()`
16631667 let claimed = self
16641668 .db
16651669 .prepare(
1666− "UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, started_at = ?, seen_at = ? WHERE id = ? AND status = 'queued' RETURNING id",
1670+ "UPDATE jobs SET status = 'in_progress', token_hash = ?, first_token_hash = NULL, started_at = ?, seen_at = ?, waiting_since = NULL WHERE id = ? AND status = 'queued' RETURNING id",
16671671 )
16681672 .bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
16691673 .first::<Value>(None)
17031707 .await
17041708 .unwrap_or_else(|error| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
17051709 if let Outcome::Fail(refused) = started {
1710+ // No machine free (every sandbox of its class taken, or a
1711+ // rollout replacing them): back in the queue, tried again at
1712+ // the next sweep, until it has waited WAIT_FOR_MACHINE_MS.
1713+ // The sandboxes after it would find none either, so this
1714+ // sweep starts no more.
1715+ if refused.code == FailureCode::Unavailable {
1716+ let since = job.waiting_since.clone().unwrap_or_else(|| at.clone());
1717+ if crate::waited_for_a_machine(Some(&since), now_ms()) {
1718+ let minutes = crate::WAIT_FOR_MACHINE_MS / 60_000;
1719+ let reason = format!("{} None came free in {minutes} minutes.", refused.message);
1720+ Box::pin(self.finish_job(&job.id, "failure", Some(&reason), None)).await?;
1721+ continue;
1722+ }
1723+ self.db
1724+ .prepare(
1725+ "UPDATE jobs SET status = 'queued', token_hash = NULL, started_at = NULL, seen_at = NULL, waiting_since = ?, reason = ?
1726+ WHERE id = ? AND status = 'in_progress'",
1727+ )
1728+ .bind(&[since.as_str().into(), refused.message.as_str().into(), job.id.as_str().into()])?
1729+ .run()
1730+ .await?;
1731+ worker::console_warn!("actions: job {} waits for a machine: {}", job.id, refused.message);
1732+ break;
1733+ }
17061734 // Billing would not pay for a scheduled run's job: the
17071735 // schedule waits rather than making runs only to refuse them.
17081736 if refused.code == FailureCode::PaymentRequired
+6−1
3636 import { BACKUP_MINUTES, backupEnv, backupPace, backupSandboxName } from "./backup";
3737 import { holdCredentials, revokeCredentials, runCredential } from "./credentials";
3838 import { cancelTask, enqueueTask, handedOverStep, selfHostedRoute, taskEnv, taskRepo } from "./self-hosted";
39−import { answered, capReached, describeError, type Owed, recordOwed, stoppedShort, tellStopped, withinTimeCap } from "./lifecycle";
39+import { answered, capReached, describeError, noMachine, type Owed, recordOwed, stoppedShort, tellStopped, withinTimeCap } from "./lifecycle";
4040 import {
4141 ABUSE_EXIT_CODE,
4242 ABUSE_HOST,
996996 },
997997 });
998998 } catch (error) {
999+ // No machine free: the actions service puts the job back in the
1000+ // queue for its next sweep. What was reserved is given back already.
1001+ if (noMachine(error)) {
1002+ return { ok: false, error: { code: "unavailable", message: `No ${instance.label} machine was free to start the job.` } };
1003+ }
9991004 // A sandbox that could not start, or stopped at once: the job fails
10001005 // with why, rather than waiting to be noticed.
10011006 return {
+8−0
55 answered,
66 capReached,
77 describeError,
8+ noMachine,
89 OWED_GIVE_UP_MS,
910 OWED_MAX_DELAY_SECONDS,
1011 owedDelaySeconds,
169170 assert.equal(owedDelaySeconds(8), OWED_MAX_DELAY_SECONDS);
170171 assert.equal(owedDelaySeconds(1000), OWED_MAX_DELAY_SECONDS);
171172 });
173+
174+test("a sandbox Cloudflare has no container for waits for a machine; any other failure does not", () => {
175+ assert.equal(noMachine(new Error("there is no container instance that can be provided to this durable object")), true);
176+ assert.equal(noMachine("Error: There is no Container instance available at this time"), true);
177+ assert.equal(noMachine(new Error("the container exited with code 1")), false);
178+ assert.equal(noMachine(undefined), false);
179+});
+9−0
8484 return exitCode !== 0 || why !== null;
8585 }
8686
87+/**
88+ * Whether a sandbox could not start because Cloudflare had no container to
89+ * give it: every instance of its class is taken, or a rollout is replacing
90+ * them. It passes; the work waits for a machine rather than failing.
91+ */
92+export function noMachine(error: unknown): boolean {
93+ return /no container instance/i.test(describeError(error));
94+}
95+
8796 /** An error as one line, for a log: its message, or what was thrown. */
8897 export function describeError(error: unknown): string {
8998 if (error instanceof Error) return error.message || error.name;