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

Queued jobs start after a request answers, so the caller leaving cannot cancel their start #185

Mergedsyntaqx merged start-after-answer into main
2 files+40−40/2 viewed
+34−3
185185 limits: limits::Limits,
186186 /// Whether repositories' Actions are on, read once per request.
187187 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>,
188196 }
189197
190198 impl Actions {
205213 blob_key: env.secret("ACTIONS_KEY").ok().map(|key| runtime::blob_key(&key.to_string())),
206214 limits: limits::Limits::from_env(env),
207215 actions_on: protection::ActionsOn::default(),
216+ defer_starts: false,
217+ starts_wanted: std::cell::Cell::new(false),
208218 })
209219 }
210220
284294 }
285295
286296 #[event(fetch)]
287−async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
297+async fn fetch(mut request: Request, env: Env, ctx: Context) -> Result<Response> {
288298 let Some(method) = rpc_method(&request) else {
289299 return Response::error("Not found", 404);
290300 };
291301 let body: Value = request.json().await?;
292− let service = Actions::new(&env)?;
293− match method.as_str() {
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.
323+async fn answer(service: &Actions, method: &str, body: Value) -> Result<Response> {
324+ match method {
294325 "workflows" => reply(&service.workflows(args(body)?).await?),
295326 "workflow" => reply(&service.workflow(args(body)?).await?),
296327 "runs" => reply(&service.runs(args(body)?).await?),
+6−1
15971597 Ok(())
15981598 }
15991599
1600− /// Starts queued jobs, oldest first, while their workspace has room.
1600+ /// Starts queued jobs, oldest first, while their workspace has room. In
1601+ /// a request, only notes that they want starting (`defer_starts`).
16011602 pub async fn start_queued(&self) -> Result<()> {
1603+ if self.defer_starts {
1604+ self.starts_wanted.set(true);
1605+ return Ok(());
1606+ }
16021607 self.start_queued_except(&std::collections::HashSet::new()).await
16031608 }
16041609