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
531 linesCodeBlameRaw
1//! The actions service: a repository's GitHub Actions workflows, run on
2//! g1t as they are. See `g1t_contracts::actions` for the methods and
3//! `g1t_actions` for how workflows, expressions and filters are read.
4//!
5//! - [`sync`] reads workflow files, at the default branch for the list and
6//! at an event's own commit for its runs.
7//! - [`trigger`] turns events, schedules and manual runs into runs.
8//! - [`plan`] moves a run's jobs along: each waits for the jobs it needs,
9//! is skipped or expanded into its matrix, queued, started in a sandbox
10//! when the workspace has room, and finished by the sandbox's report.
11//! - [`payload`] builds the webhook-shaped `github.event`.
12//! - [`settings`] keeps secrets and variables, and [`broker`] the stand-ins
13//! jobs are given for brokered secrets.
14//! - [`protection`] keeps a repository's workflow settings and its
15//! environments' protection rules, holds jobs at those rules until they
16//! pass, and lets pull requests' runs from outside wait for approval.
17//! - [`cache`] lists `actions/cache` entries, which the API keeps in R2.
18//! - [`reach`] finds the actions and reusable workflows of other
19//! repositories on g1t, and [`registry`] resolves those from GitHub to
20//! commits and keeps their archives, for jobs to download from g1t.
21//! - [`runners`] keeps self-hosted runners, hands them jobs (and agent
22//! work from the runner service) when they ask, and hears back.
23//!
24//! The service acts as the repository's workspace: it reads what the
25//! workspace can read, and a job's `GITHUB_TOKEN` is a token of the
26//! workspace's that reaches the job's repository only, with the scopes its
27//! `permissions:` give it, revoked when the job ends.
28
29mod run_artifacts;
30mod broker;
31mod cache;
32mod limits;
33mod managed;
34mod payload;
35mod plan;
36mod protection;
37mod reach;
38mod registry;
39mod rename;
40mod retention;
41pub mod runtime;
42mod runners;
43mod settings;
44mod sync;
45mod trigger;
46mod views;
47
48use g1t_contracts::access::{self, Capability};
49use g1t_contracts::events::Event;
50use g1t_contracts::identity::{SlugArgs, Workspace};
51use g1t_contracts::repos::{GetArgs, GetByIdArgs, Repo, RepoPath};
52use g1t_contracts::{FailureCode, Membership, Outcome, PrincipalKind, User, Viewer};
53use g1t_kit::{args, reply, rpc_method};
54use g1t_secrets::Sealer;
55use serde::Deserialize;
56use serde_json::Value;
57use worker::wasm_bindgen::JsValue;
58use worker::{Context, D1Database, Env, Fetcher, MessageBatch, MessageExt, Request, Response, Result, ScheduleContext, ScheduledEvent, event};
59
60/// The most workflow files read from a repository.
61pub const MAX_WORKFLOWS: usize = 50;
62/// Jobs on g1t's own machines one workspace may always have running at
63/// once; the rest wait their turn, unless the pool has room (`has_room`).
64pub const RUNNING_PER_WORKSPACE: u32 = 4;
65/// The most one workspace may run at once while the pool has room to spare.
66pub const BURST_PER_WORKSPACE: u32 = 12;
67/// Sandboxes g1t's own machines have between them (services/runner's
68/// wrangler.jsonc: 20 standard, 10 g1t-2core, 30 g1t-4core, which every
69/// public repository's ubuntu-latest jobs run on).
70pub const HOSTED_POOL: u32 = 60;
71/// Kept free while workspaces run past their share, so another workspace's
72/// first jobs still start at once.
73pub const POOL_HEADROOM: u32 = 8;
74
75/// Whether a workspace running `in_workspace` jobs on g1t's machines may
76/// start another while `in_pool` run across every workspace: always within
77/// its share, and past it, up to the burst, while the pool keeps its
78/// headroom.
79pub fn has_room(in_workspace: u32, in_pool: u32) -> bool {
80 in_workspace < RUNNING_PER_WORKSPACE
81 || (in_workspace < BURST_PER_WORKSPACE && in_pool + POOL_HEADROOM < HOSTED_POOL)
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.
87pub const WAIT_FOR_MACHINE_MS: u64 = 30 * 60 * 1000;
88
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.
97pub 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.
102pub 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
111/// Whether a job that has waited for a machine since `since` (RFC 3339)
112/// should stop waiting at `now`. One whose start cannot be read waits from
113/// now on.
114pub fn waited_for_a_machine(since: Option<&str>, now: u64) -> bool {
115 since
116 .and_then(g1t_contracts::time::parse_rfc3339)
117 .is_some_and(|since| now.saturating_sub(since) >= WAIT_FOR_MACHINE_MS)
118}
119/// The longest a job on g1t's own machines may run, whatever its
120/// `timeout-minutes`.
121pub const MAX_TIMEOUT_MINUTES: u32 = g1t_actions::workflow::HOSTED_MAX_TIMEOUT_MINUTES;
122/// The longest a job on a self-hosted runner may run: the machine is the
123/// workspace's own, and its time costs nothing.
124pub const SELF_HOSTED_MAX_TIMEOUT_MINUTES: u32 = 24 * 60;
125/// A running job that has said nothing for this long is taken as lost.
126pub const SILENT_MS: u64 = 10 * 60 * 1000;
127/// Schedules pause in a repository that has had no push for this long; the
128/// next push resumes them.
129pub const SCHEDULE_IDLE_MS: u64 = 60 * 24 * 60 * 60 * 1000;
130/// How long a workflow's schedule waits after billing refused to start a
131/// job of one of its scheduled runs, before it tries again.
132pub const SCHEDULE_REFUSED_MS: u64 = 60 * 60 * 1000;
133pub const SITE: &str = "https://g1t.sh";
134pub const API: &str = "https://api.g1t.sh";
135
136#[derive(Deserialize)]
137struct Count {
138 n: u32,
139}
140
141pub fn optional(value: Option<&str>) -> JsValue {
142 value.map_or(JsValue::NULL, JsValue::from)
143}
144
145pub fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> {
146 Outcome::fail(code, message)
147}
148
149/// `owner/name` as a path.
150pub fn repo_path(full_name: &str) -> RepoPath {
151 let (namespace, name) = full_name.split_once('/').unwrap_or((full_name, ""));
152 RepoPath {
153 namespace: namespace.to_owned(),
154 name: name.to_owned(),
155 }
156}
157
158pub struct Actions {
159 db: D1Database,
160 repos: Fetcher,
161 work: Fetcher,
162 identity: Fetcher,
163 runner: Fetcher,
164 events: Fetcher,
165 /// Projects: a repository's secrets and variables belong to its project.
166 projects: Fetcher,
167 /// Billing: self-hosted runners' time, recorded at $0, and the
168 /// cache's storage.
169 billing: Fetcher,
170 /// Deployments: a job with an `environment:` deploys to it, and its
171 /// run's deployment is recorded there (plan.rs `report_deployment`).
172 deployments: Fetcher,
173 /// Notify, optional: pages open on a repository hear its runs' jobs
174 /// start and finish (plan.rs `tell_live`). Without it they catch up
175 /// on their own.
176 notify: Option<Fetcher>,
177 /// Where the API keeps cache entries, for deleting evicted ones.
178 cache: Option<worker::Bucket>,
179 /// Seals secrets; absent until `ACTIONS_KEY` is set, when secrets
180 /// cannot be saved.
181 sealer: Option<Sealer>,
182 /// Signs the toolkit's blob links (runtime.rs), from `ACTIONS_KEY`.
183 blob_key: Option<String>,
184 /// Log retention and free workspaces' caps, from wrangler vars (limits.rs).
185 limits: limits::Limits,
186 /// Whether repositories' Actions are on, read once per request.
187 actions_on: protection::ActionsOn,
188 /// In a request, queued jobs are not started inline: whoever made the
189 /// request may be gone before a sandbox starts, which can take seconds
190 /// while Cloudflare has no container free, and the start would be
191 /// cancelled with it, after the jobs were marked as running. They are
192 /// started after the answer instead (`fetch`, `wait_until`).
193 defer_starts: bool,
194 /// Whether this request queued jobs or freed room for them.
195 starts_wanted: std::cell::Cell<bool>,
196}
197
198impl Actions {
199 fn new(env: &Env) -> Result<Self> {
200 Ok(Actions {
201 db: env.d1("DB")?,
202 repos: env.service("REPOS")?,
203 work: env.service("WORK")?,
204 identity: env.service("IDENTITY")?,
205 runner: env.service("RUNNER")?,
206 events: env.service("EVENTS")?,
207 projects: env.service("PROJECTS")?,
208 billing: env.service("BILLING")?,
209 deployments: env.service("DEPLOYMENTS")?,
210 notify: env.service("NOTIFY").ok(),
211 cache: env.bucket("ACTIONS_CACHE").ok(),
212 sealer: env.secret("ACTIONS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())),
213 blob_key: env.secret("ACTIONS_KEY").ok().map(|key| runtime::blob_key(&key.to_string())),
214 limits: limits::Limits::from_env(env),
215 actions_on: protection::ActionsOn::default(),
216 defer_starts: false,
217 starts_wanted: std::cell::Cell::new(false),
218 })
219 }
220
221 /// The workspace itself, as the service acts.
222 async fn workspace_actor(&self, slug: &str) -> Result<Option<User>> {
223 let workspace: Option<Workspace> = g1t_kit::call(&self.identity, "get_workspace", &SlugArgs { slug: slug.to_owned() }).await?;
224 Ok(workspace.map(|workspace| User {
225 id: workspace.id,
226 username: workspace.slug.clone(),
227 kind: PrincipalKind::Workspace,
228 verified: true,
229 workspaces: vec![Membership::member(workspace.slug)],
230 ..User::default()
231 }))
232 }
233
234 /// The repository, if the viewer may see it and it is not a pull
235 /// request's working copy.
236 async fn visible_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Option<Repo>> {
237 let found: Outcome<Repo> = g1t_kit::call(
238 &self.repos,
239 "get",
240 &GetArgs {
241 path: path.clone(),
242 viewer: viewer.clone(),
243 },
244 )
245 .await?;
246 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()))
247 }
248
249 /// A repository by id, as its workspace sees it.
250 async fn repo_by_id(&self, id: &str) -> Result<Option<(Repo, User)>> {
251 let path: Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: id.to_owned() }).await?;
252 let Some(path) = path else { return Ok(None) };
253 let Some(actor) = self.workspace_actor(&path.namespace).await? else {
254 return Ok(None);
255 };
256 let found: Outcome<Repo> = g1t_kit::call(
257 &self.repos,
258 "get_by_id",
259 &GetByIdArgs {
260 id: id.to_owned(),
261 viewer: Some(actor.clone()),
262 },
263 )
264 .await?;
265 Ok(found.into_result().ok().filter(|repo| repo.fork_of.is_none()).map(|repo| (repo, actor)))
266 }
267
268 /// The repository at `path`, when `actor` may do `capability` in it:
269 /// not found when they cannot read it, forbidden when their role falls
270 /// short.
271 async fn may(&self, actor: &User, path: &RepoPath, capability: Capability) -> Result<Outcome<Repo>> {
272 let Some(repo) = self.visible_repo(path, &Some(actor.clone())).await? else {
273 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
274 };
275 if !access::can(Some(actor), &repo, capability) {
276 return Ok(fail(
277 FailureCode::Forbidden,
278 access::needs(capability, &format!("{}/{}", repo.namespace, repo.name)),
279 ));
280 }
281 Ok(Outcome::Ok(repo))
282 }
283}
284
285/// Unwraps an `Outcome`, or returns its failure from the enclosing method.
286#[macro_export]
287macro_rules! check {
288 ($outcome:expr) => {
289 match $outcome {
290 g1t_contracts::Outcome::Ok(value) => value,
291 g1t_contracts::Outcome::Fail(refused) => return Ok(g1t_contracts::Outcome::Fail(refused)),
292 }
293 };
294}
295
296#[event(fetch)]
297async fn fetch(mut request: Request, env: Env, ctx: Context) -> Result<Response> {
298 let Some(method) = rpc_method(&request) else {
299 return Response::error("Not found", 404);
300 };
301 let body: Value = request.json().await?;
302 let service = Actions { defer_starts: true, ..Actions::new(&env)? };
303 let response = answer(&service, &method, body).await;
304 // The jobs this request queued or made room for start after the
305 // answer, where the caller leaving cannot cancel them. A start that is
306 // cut short anyway is tried again (FIRST_REPORT_MS).
307 if service.starts_wanted.get() {
308 ctx.wait_until(async move {
309 match Actions::new(&env) {
310 Ok(service) => {
311 if let Err(error) = service.start_queued().await {
312 worker::console_error!("actions: starting queued jobs after {method} failed: {error}");
313 }
314 }
315 Err(error) => worker::console_error!("actions: could not start: {error}"),
316 }
317 });
318 }
319 response
320}
321
322/// One RPC's answer.
323async fn answer(service: &Actions, method: &str, body: Value) -> Result<Response> {
324 match method {
325 "workflows" => reply(&service.workflows(args(body)?).await?),
326 "workflow" => reply(&service.workflow(args(body)?).await?),
327 "runs" => reply(&service.runs(args(body)?).await?),
328 "run" => reply(&service.run(args(body)?).await?),
329 "logs" => reply(&service.logs(args(body)?).await?),
330 "job_log_text" => reply(&service.job_log_text(args(body)?).await?),
331 "run_logs" => reply(&service.run_logs(args(body)?).await?),
332 "summaries" => reply(&service.summaries(args(body)?).await?),
333 "dispatch" => reply(&service.dispatch(args(body)?).await?),
334 "repository_dispatch" => reply(&service.repository_dispatch(args(body)?).await?),
335 "approve_run" => reply(&service.approve_run(args(body)?).await?),
336 "pending_deployments" => reply(&service.pending_deployments(args(body)?).await?),
337 "review_deployments" => reply(&service.review_deployments(args(body)?).await?),
338 "actions_settings" => reply(&service.actions_settings(args(body)?).await?),
339 "set_actions_settings" => reply(&service.set_actions_settings(args(body)?).await?),
340 "workspace_actions_settings" => reply(&service.workspace_actions_settings(args(body)?).await?),
341 "set_workspace_actions_settings" => reply(&service.set_workspace_actions_settings(args(body)?).await?),
342 "environments" => reply(&service.environments(args(body)?).await?),
343 "set_environment" => reply(&service.set_environment(args(body)?).await?),
344 "delete_environment" => reply(&service.delete_environment(args(body)?).await?),
345 "merge_group" => reply(&service.merge_group(args(body)?).await?),
346 "cancel" => reply(&service.cancel(args(body)?).await?),
347 "rerun" => reply(&service.rerun(args(body)?).await?),
348 "set_workflow_enabled" => reply(&service.set_workflow_enabled(args(body)?).await?),
349 "settings" => reply(&service.settings(args(body)?).await?),
350 "set_setting" => reply(&service.set_setting(args(body)?).await?),
351 "delete_setting" => reply(&service.delete_setting(args(body)?).await?),
352 "resolve_settings" => reply(&service.resolve_settings(args(body)?).await?),
353 // Managed workflows: g1t's services' side only.
354 "set_managed_workflow" => reply(&service.set_managed_workflow(args(body)?).await?),
355 "remove_managed_workflow" => reply(&service.remove_managed_workflow(args(body)?).await?),
356 "set_managed_deploy" => reply(&service.set_managed_deploy(args(body)?).await?),
357 // A workspace's managed workflows: its owners'.
358 "workspace_managed_workflows" => reply(&service.workspace_managed_workflows(args(body)?).await?),
359 "set_workspace_managed_workflow" => reply(&service.set_workspace_managed_workflow(args(body)?).await?),
360 "remove_workspace_managed_workflow" => reply(&service.remove_workspace_managed_workflow(args(body)?).await?),
361 "job_spec" => reply(&service.job_spec(args(body)?).await?),
362 "job_auth" => reply(&service.job_auth(args(body)?).await?),
363 "job_action" => reply(&service.job_action(args(body)?).await?),
364 "job_action_archive" => reply(&service.job_action_archive(args(body)?).await?),
365 "check_runs" => reply(&service.check_runs(args(body)?).await?),
366 "job_report" => reply(&service.job_report(args(body)?).await?),
367 // A running job's stand-ins for brokered secrets: the runner
368 // Worker's alone, never passed on by the API.
369 "brokered_secrets" => reply(&service.brokered_secrets(args(body)?).await?),
370 // The same for a self-hosted job's broker requests, which the
371 // runner Worker relays (`POST /actions/jobs/{job}/broker`): with
372 // the job's current token only.
373 "relayed_secrets" => reply(&service.relayed_secrets(args(body)?).await?),
374 // actions/cache, through the API with the job's token.
375 "cache_lookup" => reply(&service.cache_lookup(args(body)?).await?),
376 "cache_reserve" => reply(&service.cache_reserve(args(body)?).await?),
377 "cache_commit" => reply(&service.cache_commit(args(body)?).await?),
378 "cache_abort" => reply(&service.cache_abort(args(body)?).await?),
379 "cache_upload" => reply(&service.cache_upload(args(body)?).await?),
380 // The cache: people's side, listing what is saved and deleting it.
381 "caches" => reply(&service.caches(args(body)?).await?),
382 "cache_usage" => reply(&service.cache_usage(args(body)?).await?),
383 "delete_caches" => reply(&service.delete_caches(args(body)?).await?),
384 // Artifacts: a job's side, with its token or its runtime token.
385 "artifact_reserve" => reply(&service.artifact_reserve(args(body)?).await?),
386 "artifact_commit" => reply(&service.artifact_commit(args(body)?).await?),
387 "artifact_abort" => reply(&service.artifact_abort(args(body)?).await?),
388 "job_artifacts" => reply(&service.job_artifacts(args(body)?).await?),
389 "job_artifact" => reply(&service.job_artifact(args(body)?).await?),
390 "job_delete_artifact" => reply(&service.job_delete_artifact(args(body)?).await?),
391 // Artifacts: people's side.
392 "artifacts" => reply(&service.artifacts(args(body)?).await?),
393 "artifact" => reply(&service.artifact(args(body)?).await?),
394 "artifact_download" => reply(&service.artifact_download(args(body)?).await?),
395 "delete_artifact" => reply(&service.delete_artifact(args(body)?).await?),
396 "artifact_retention" => reply(&service.artifact_retention(args(body)?).await?),
397 // The toolkit's protocols: the runtime token, OIDC, blob links.
398 "runtime_auth" => reply(&service.runtime_auth(args(body)?).await?),
399 "oidc_claims" => reply(&service.oidc_claims(args(body)?).await?),
400 "blob_sign" => reply(&service.blob_sign(args(body)?).await?),
401 "blob_open" => reply(&service.blob_open(args(body)?).await?),
402 "blob_part" => reply(&service.blob_part(args(body)?).await?),
403 "blob_parts" => reply(&service.blob_parts(args(body)?).await?),
404 "blob_done" => reply(&service.blob_done(args(body)?).await?),
405 // Self-hosted runners: people's side.
406 "runners" => reply(&service.runners(args(body)?).await?),
407 "create_registration_token" => reply(&service.create_registration_token(args(body)?).await?),
408 "remove_runner" => reply(&service.remove_runner(args(body)?).await?),
409 // Staff only, from sudo: a workspace's runners, and removing them.
410 "admin_runners" => reply(&service.admin_runners(args(body)?).await?),
411 "admin_remove_runners" => reply(&service.admin_remove_runners(args(body)?).await?),
412 "runner_groups" => reply(&service.runner_groups(args(body)?).await?),
413 "set_runner_group" => reply(&service.set_runner_group(args(body)?).await?),
414 "delete_runner_group" => reply(&service.delete_runner_group(args(body)?).await?),
415 "runner_settings" => reply(&service.runner_settings(args(body)?).await?),
416 "set_runner_settings" => reply(&service.set_runner_settings(args(body)?).await?),
417 // The desktop app's runners, as each app says they are.
418 "report_desktop" => reply(&service.report_desktop(args(body)?).await?),
419 "desktop_runners" => reply(&service.desktop_runners(args(body)?).await?),
420 // The runner's own side, through the API with its credential.
421 "runner_register" => reply(&service.runner_register(args(body)?).await?),
422 "runner_poll" => reply(&service.runner_poll(args(body)?).await?),
423 "runner_finished" => reply(&service.runner_finished(args(body)?).await?),
424 "runner_remove_self" => reply(&service.runner_remove_self(args(body)?).await?),
425 // Agent work, from the runner service.
426 "runner_route" => reply(&service.runner_route(args(body)?).await?),
427 "stuck_jobs" => reply(&service.stuck_jobs(args(body)?).await?),
428 "runner_activity" => reply(&service.runner_activity(args(body)?).await?),
429 "enqueue_task" => reply(&service.enqueue_task(args(body)?).await?),
430 "cancel_task" => reply(&service.cancel_task(args(body)?).await?),
431 _ => Response::error("Unknown method", 404),
432 }
433}
434
435/// Events from the bus, on this service's own queue.
436#[event(queue)]
437async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
438 let service = Actions::new(&env)?;
439 for message in batch.messages()? {
440 // A workspace renamed: its rows move to the slug it has now.
441 if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), rename::STATEMENTS).await? {
442 message.ack();
443 continue;
444 }
445 // A repository renamed or transferred: its rows follow its new path.
446 if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), rename::TRANSFERRED).await? {
447 message.ack();
448 continue;
449 }
450 // A workspace deleted: what it kept for itself goes.
451 if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), rename::DELETED).await? {
452 message.ack();
453 continue;
454 }
455 // A repository purged: every row kept for it goes.
456 if g1t_kit::lifecycle::on_purged(&env.d1("DB")?, message.body(), rename::PURGED).await? {
457 message.ack();
458 continue;
459 }
460 // A repository deleted or archived: its runs stop.
461 if let Some(repo_id) = plan::stops_runs(message.body()) {
462 if let Err(error) = service.stop_runs(&repo_id).await {
463 worker::console_error!("actions: event {} failed: {error}", message.body().id);
464 message.retry();
465 continue;
466 }
467 message.ack();
468 continue;
469 }
470 if let Err(error) = service.on_event(message.body()).await {
471 worker::console_error!("actions: event {} failed: {error}", message.body().id);
472 message.retry();
473 continue;
474 }
475 message.ack();
476 }
477 Ok(())
478}
479
480/// Every minute: schedules that fire, jobs waiting for room, and jobs
481/// whose sandbox went quiet.
482#[event(scheduled)]
483async fn scheduled(_event: ScheduledEvent, env: Env, _ctx: ScheduleContext) {
484 match Actions::new(&env) {
485 Ok(service) => {
486 if let Err(error) = service.on_minute(g1t_kit::now_ms()).await {
487 worker::console_error!("actions: the sweep failed: {error}");
488 }
489 }
490 Err(error) => worker::console_error!("actions: could not start: {error}"),
491 }
492}
493
494#[cfg(test)]
495mod room_tests {
496 use super::*;
497
498 #[test]
499 fn a_workspace_always_has_its_share() {
500 assert!(has_room(RUNNING_PER_WORKSPACE - 1, HOSTED_POOL));
501 assert!(!has_room(RUNNING_PER_WORKSPACE, HOSTED_POOL));
502 }
503
504 #[test]
505 fn past_its_share_a_workspace_runs_more_while_the_pool_keeps_its_headroom() {
506 assert!(has_room(RUNNING_PER_WORKSPACE, HOSTED_POOL - POOL_HEADROOM - 1));
507 assert!(!has_room(RUNNING_PER_WORKSPACE, HOSTED_POOL - POOL_HEADROOM));
508 assert!(!has_room(BURST_PER_WORKSPACE, 0));
509 }
510
511 #[test]
512 fn a_job_waits_for_a_machine_for_half_an_hour() {
513 let since = g1t_contracts::time::rfc3339(1_000_000);
514 assert!(!waited_for_a_machine(None, u64::MAX));
515 assert!(!waited_for_a_machine(Some(&since), 1_000_000 + WAIT_FOR_MACHINE_MS - 1));
516 assert!(waited_for_a_machine(Some(&since), 1_000_000 + WAIT_FOR_MACHINE_MS));
517 assert!(!waited_for_a_machine(Some("not a time"), u64::MAX));
518 }
519
520 #[test]
521 fn a_job_that_never_reported_is_lost_after_three_minutes() {
522 let started = g1t_contracts::time::rfc3339(1_000_000);
523 let later = g1t_contracts::time::rfc3339(1_005_000);
524 assert!(!never_reported(Some(&started), Some(&started), 1_000_000 + FIRST_REPORT_MS - 1));
525 assert!(never_reported(Some(&started), Some(&started), 1_000_000 + FIRST_REPORT_MS));
526 // Once it has said anything, only SILENT_MS ends it.
527 assert!(!never_reported(Some(&started), Some(&later), u64::MAX));
528 assert!(!never_reported(None, None, u64::MAX));
529 assert!(!never_reported(Some("not a time"), Some("not a time"), u64::MAX));
530 }
531}