Merge pull request #181 from flagon-io/wait-for-a-machine

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;

80	80		in_workspace < RUNNING_PER_WORKSPACE
81	81		\|\| (in_workspace < BURST_PER_WORKSPACE && in_pool + POOL_HEADROOM < HOSTED_POOL)
82	82		}
	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	+	}
83	97		/// The longest a job on g1t's own machines may run, whatever its
84	98		/// `timeout-minutes`.
85	99		pub const MAX_TIMEOUT_MINUTES: u32 = g1t_actions::workflow::HOSTED_MAX_TIMEOUT_MINUTES;
⋯
440	454		assert!(!has_room(RUNNING_PER_WORKSPACE, HOSTED_POOL - POOL_HEADROOM));
441	455		assert!(!has_room(BURST_PER_WORKSPACE, 0));
442	456		}
	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	+	}
443	466		}

253	253		/// for another (migration 0020): good for `brokered_secrets` only.
254	254		#[serde(default)]
255	255		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>,
256	260		}
257	261
258	262		/// How long a cancelled job has to run its `if: always()` and `cancelled()`
⋯
1663	1667		let claimed = self
1664	1668		.db
1665	1669		.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",
1667	1671		)
1668	1672		.bind(&[sha256_hex(&token).into(), at.as_str().into(), at.as_str().into(), job.id.as_str().into()])?
1669	1673		.first::<Value>(None)
⋯
1703	1707		.await
1704	1708		.unwrap_or_else(\|error\| fail(FailureCode::Conflict, format!("The runner could not be reached: {error}")));
1705	1709		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	+	}
1706	1734		// Billing would not pay for a scheduled run's job: the
1707	1735		// schedule waits rather than making runs only to refuse them.
1708	1736		if refused.code == FailureCode::PaymentRequired

36	36		import { BACKUP_MINUTES, backupEnv, backupPace, backupSandboxName } from "./backup";
37	37		import { holdCredentials, revokeCredentials, runCredential } from "./credentials";
38	38		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";
40	40		import {
41	41		ABUSE_EXIT_CODE,
42	42		ABUSE_HOST,
⋯
996	996		},
997	997		});
998	998		} 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	+	}
999	1004		// A sandbox that could not start, or stopped at once: the job fails
1000	1005		// with why, rather than waiting to be noticed.
1001	1006		return {

5	5		answered,
6	6		capReached,
7	7		describeError,
	8	+	noMachine,
8	9		OWED_GIVE_UP_MS,
9	10		OWED_MAX_DELAY_SECONDS,
10	11		owedDelaySeconds,
⋯
169	170		assert.equal(owedDelaySeconds(8), OWED_MAX_DELAY_SECONDS);
170	171		assert.equal(owedDelaySeconds(1000), OWED_MAX_DELAY_SECONDS);
171	172		});
	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	+	});

84	84		return exitCode !== 0 \|\| why !== null;
85	85		}
86	86
	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	+
87	96		/** An error as one line, for a log: its message, or what was thrown. */
88	97		export function describeError(error: unknown): string {
89	98		if (error instanceof Error) return error.message \|\| error.name;

flagon-io/g1t

Comments