GITGUD

neiam neiam /gitgud

Git Gud

public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages

⭐ 1

Log in to mark this repository.

Actions: cancel reaches the runner, and concurrency: groups are honoured

6b5083e · Gabriel Morell · 2026-10-17 16:18

51 files +1819 -207
Message 
{commit_body(@commit)}

Files changed

modified README.md
+5 −4
@@ -93,7 +93,8 @@ reinvent them.
93 93 - **`paths:` / `paths-ignore:`** with GitHub's pattern rules (`*` stays within a directory, `**` crosses them, `?`, `[a-z]`; in `paths:`, `!pattern` un-matches and a later positive pattern re-matches). A push runs if any changed file is selected by `paths:`, or if not every changed file is matched by `paths-ignore:`; both set → `paths:` wins, with a warning. Changed files are the ref's old → new diff (renames count as both paths); a new branch is diffed from where it left the default branch. Tag pushes and manual runs ignore path filters, and when the changed files can't be worked out (no default branch to compare with, an unknown base, a git error, more than 3000 files) the filter passes — a run is never skipped because the forge couldn't look
94 94 - **The job's context** follows Forgejo's: `github.event_name`, `ref`, `ref_name`, `ref_type`, `sha`, `repository` (`owner/name`), `repository_id`, `repository_owner(_id)`, `actor(_id)`, `run_id`, `run_number`, `run_attempt` (1, bumped by re-running the job; re-running the whole run makes a new run and a new `run_id`), `workflow`, `workflow_ref` (`owner/name/<path>@<ref>`), `job`, `server_url`, `api_url`, `token`. **`github.event`** is GitHub's webhook payload, trimmed: every event has `repository` and `sender`; `push` has `ref`, `before` (40 zeros for a new ref), `after`, `created`, `deleted`, `forced` (always `false`: not tracked), `compare`, `commits` (at most 20) and `head_commit`, `pusher`; `pull_request` has `action`, `number` and `pull_request` (`title`, `body`, `html_url`, `user`, `head`/`base` with `ref`, `sha`, `label`, `repo`); `workflow_dispatch` has `inputs`, `ref`, `workflow`. **`secrets.GITHUB_TOKEN`** (and `GITEA_TOKEN`, `FORGEJO_TOKEN`) is the job token. The workflow's top-level `env:`, `defaults:`, `permissions:`, `run-name:` and `on:` reach the runner with the job
95 95 - **Jobs that depend on jobs**, as on GitHub: `${{ needs.<id>.outputs.* }}` and `needs.<id>.result` (`success`, `failure`, `cancelled`, `skipped`; for a matrix, the worst leg's, with the legs' outputs merged) reach the dependent job. A job waits until all its needs finish; if any didn't succeed it is **skipped** — and its own dependents with it — unless its `if:` calls `always()`, `failure()` or `cancelled()`, so cleanup and "report the failure" jobs run. A job its own `if:` skipped shows as skipped, not green. Job-level `continue-on-error: true` (or `${{ matrix.* }}` that comes out `true`) keeps a failure from skipping its dependents, failing the run or fail-fasting its matrix siblings, though the job still shows as failed; any other expression there counts as false. Skipped jobs don't fail a run. Unlike GitHub, **cancelling a run cancels its waiting jobs too**, `always()` or not
96 − - Re-run / cancel buttons on the workflow show page
96 + - Re-run / cancel buttons on the workflow show page. **Cancelling stops running jobs**: the runner hears it in the reply to its next progress report and stops the job, still uploading its log; a cancelled job stays cancelled whatever its runner reports afterwards
97 + - **`concurrency:`**, workflow and job level, as a group name or `{group:, cancel-in-progress:}`: within a group (per repository) one run or job is in progress and at most one waits (shown as "Waiting for <group>"); a newer arrival cancels the waiting one, or with `cancel-in-progress` the one in progress too, and when the one in progress finishes the waiting one starts. A run (or job) cancelled only by a newer one in its group concludes `cancelled`, not failed. A workflow's group is decided when the run is created, a job's when it would be queued, so it can read `matrix` and `needs`. Group and `cancel-in-progress` may use `${{ }}` with the forge's own expression subset: property paths into `github`, `inputs`, `vars`, `matrix`, `needs` (`a.b`, `a['b']`), string/number/`true`/`false`/`null` literals, `||`, `&&`, `==`, `!=`, `!`, parentheses and `format()`. A group outside that subset (any other function, say) is ignored with a warning, shown on the job — the run goes ahead **without** concurrency control rather than being held on a guess. Re-running a single job puts it back through its job-level group, not its run's
97 98 - **The job token** (`${{ github.token }}`, an HS256 JWT minted per job on `FetchTask`) works only while its job runs and for five minutes after (for the runner's post-job cache save and artifact upload), and only for that attempt of the job — a leaked token no longer stays valid for a day. It clones and fetches its own repository, reads and publishes its packages, reads its metadata over the API, dispatches workflows (its own repository's, or one that granted it), and reaches the artifact + cache endpoints of its run. **`permissions:`** narrows it as on GitHub: workflow level, replaced by a job's own; `read-all`, `write-all`, `{}` or a map in which unnamed scopes are `none`. `contents` (`read` fetch, `write` also push — to its own repository, subject to branch protection like any direct push, attributed to whoever started the run), `packages` (`read` / `write`) and `actions` (`write` to dispatch) mean something; `metadata` is always read; the others (`issues`, `pull-requests`, `statuses`, `checks`, `deployments`, `id-token`, …) are accepted and ignored, since the token can't do those things here anyway. With no `permissions:` the token keeps `contents: read, packages: write, actions: write` — GitHub's permissive default would also let it push. **A push made with the job token starts no workflows** (no `push` run, no pull request `synchronize`), as on GitHub, so a workflow that pushes can't trigger itself forever; webhooks and pull request heads still update
98 99 - **GitHub Actions cache server v1**, scoped to the repository. Restore order: the primary key exactly, then each fallback key as a prefix (newest first), each tried on the run's own ref, then — for a `pull_request` run — the PR's base branch, then the default branch; a save goes to the run's own ref. Upload chunks are placed at their `Content-Range` offsets (`actions/cache` sends four at once) and assembled on finalize; a finalized entry is immutable. Age/size eviction worker, which also clears uploads abandoned for a day. **Not reached by a default forgejo-runner**: with `cache.enabled` (its default) the runner starts its own cache server and points `ACTIONS_CACHE_URL` at it, ignoring the URL gitgud sends; jobs only use this one if the runner has `cache.enabled: false` and `ACTIONS_CACHE_URL` set in `runner.envs`. `actions/cache@v4`'s cache service v2 isn't implemented
99 100 - A simplified **v1 "pipelines" artifact API**, scoped to the run: a job creates, lists and downloads its own run's artifacts only (so jobs of one run can share files). Not yet usable from the stock actions: forgejo-runner points `ACTIONS_RUNTIME_URL` at `/api/actions_pipeline/`, which isn't mounted, and `actions/upload-artifact@v3` / `download-artifact@v3` speak a different shape of that API (chunked `PUT`s with `itemPath`, container item listing). `@v4` (the Twirp `ArtifactService` under `ACTIONS_RESULTS_URL`) isn't supported
@@ -368,9 +369,9 @@ against a matrix context pinned to that leg's single value.
368 369 `needs: build` waits for **every** leg, and a downstream job sees one
369 370 `needs.build` however many machines produced it: the worst leg's
370 371 result, and the legs' outputs merged. `fail-fast` is on by
371 −default: the first leg to fail cancels its siblings that haven't
372 −started, while one already running is left to finish rather than
373 −killed. `max-parallel` is not implemented — every leg is claimable as
372 +default: the first leg to fail cancels its siblings, running ones
373 +included, as on GitHub (a running leg's runner is told on its next
374 +report, and still uploads its log). `max-parallel` is not implemented — every leg is claimable as
374 375 soon as it exists.
375 376
376 377 ### OCI registry
modified android/app/src/main/java/org/neiam/gitgud/app/data/RepoSort.kt
+1 −1
@@ -67,7 +67,7 @@ private val byCi = compareBy<Repository> { ciRank(it.ci_status) }.then(byPushed)
67 67 private fun ciRank(status: String?): Int = when (status) {
68 68 "failure", "timed_out" -> 0
69 69 "running" -> 1
70 − "queued", "blocked", "pending" -> 2
70 + "queued", "blocked", "waiting", "pending" -> 2
71 71 "cancelled" -> 3
72 72 "success" -> 4
73 73 else -> 5
modified android/app/src/main/java/org/neiam/gitgud/app/ui/StatusChip.kt
+2 −2
@@ -42,7 +42,7 @@ fun statusColor(status: String?): Color {
42 42 return when (status) {
43 43 "success" -> theme.success
44 44 "failure", "cancelled", "timed_out" -> theme.danger
45 − "queued", "running", "pending", "blocked" -> theme.pending
45 + "queued", "running", "pending", "blocked", "waiting" -> theme.pending
46 46 "skipped" -> theme.dim
47 47 else -> theme.dim
48 48 }
@@ -53,7 +53,7 @@ fun statusIcon(status: String?): ImageVector = when (status) {
53 53 "failure", "timed_out" -> Icons.Default.Error
54 54 "cancelled" -> Icons.Default.Cancel
55 55 "running" -> Icons.Default.PlayCircle
56 − "queued", "pending", "blocked" -> Icons.Default.HourglassEmpty
56 + "queued", "pending", "blocked", "waiting" -> Icons.Default.HourglassEmpty
57 57 "skipped" -> Icons.Default.RemoveCircle
58 58 else -> Icons.AutoMirrored.Filled.HelpOutline
59 59 }
modified android/app/src/main/java/org/neiam/gitgud/app/widget/CiWidgetModel.kt
+2 −2
@@ -21,7 +21,7 @@ enum class StatusCategory { SUCCESS, FAILURE, RUNNING, QUEUED, DIM }
21 21 * a phone watching the same things show the same dots.
22 22 */
23 23 object CiWidgetModel {
24 − private val ACTIVE = setOf("queued", "running", "pending", "blocked")
24 + private val ACTIVE = setOf("queued", "running", "pending", "blocked", "waiting")
25 25
26 26 fun isActive(status: String?) = status in ACTIVE
27 27
@@ -29,7 +29,7 @@ object CiWidgetModel {
29 29 "success" -> StatusCategory.SUCCESS
30 30 "failure", "cancelled", "timed_out" -> StatusCategory.FAILURE
31 31 "running" -> StatusCategory.RUNNING
32 − "queued", "pending", "blocked" -> StatusCategory.QUEUED
32 + "queued", "pending", "blocked", "waiting" -> StatusCategory.QUEUED
33 33 else -> StatusCategory.DIM
34 34 }
35 35
modified lib/git_gud/workflows.ex
+386 −56
@@ -19,6 +19,8 @@ defmodule GitGud.Workflows do
19 19 alias GitGud.Repo
20 20 alias GitGud.Repositories
21 21 alias GitGud.Repositories.Repository
22 + alias GitGud.Workflows.Concurrency
23 + alias GitGud.Workflows.Context
22 24 alias GitGud.Workflows.EventPayload
23 25 alias GitGud.Workflows.Matrix
24 26 alias GitGud.Workflows.Ref
@@ -449,7 +451,7 @@ defmodule GitGud.Workflows do
449 451
450 452 # ── run + jobs + steps creation ──────────────────────────────────────
451 453
452 − defp create_run(%Repository{id: rid}, %{jobs: jobs} = attrs) do
454 + defp create_run(%Repository{id: rid} = repo, %{jobs: jobs} = attrs) do
453 455 Repo.transaction(fn ->
454 456 number = next_run_number(rid)
455 457
@@ -463,14 +465,137 @@ defmodule GitGud.Workflows do
463 465 |> WorkflowRun.create_changeset(run_attrs)
464 466 |> Repo.insert!()
465 467
466 − for job <- jobs, do: create_job(run, job)
468 + {run, hold} = run_concurrency(run, repo)
469 +
470 + # Every leg exists before any is queued: queueing one can cancel
471 + # another in its job-level group, and a run whose only jobs so far
472 + # are cancelled would be settled as finished half-built.
473 + legs = Enum.flat_map(jobs, &create_job(run, &1, hold))
474 +
475 + for %WorkflowJob{status: "queued"} = leg <- legs, do: enqueue_job(leg)
467 476
468 477 # Initial pending commit status so the SCM view shows the run
469 478 # is in flight before the first claim lands.
470 479 update_run_commit_status(run)
471 480
472 − run
481 + {run, hold}
473 482 end)
483 + |> case do
484 + {:ok, {run, hold}} ->
485 + # Cancelled once this run is committed, rather than from inside its
486 + # transaction: `cancel/2` releases each one's group, and that
487 + # should see this run.
488 + for id <- hold.cancel do
489 + case Repo.get(WorkflowRun, id) do
490 + %WorkflowRun{} = superseded -> cancel(superseded, reason: hold.cancel_reason)
491 + nil -> :ok
492 + end
493 + end
494 +
495 + {:ok, run}
496 +
497 + error ->
498 + error
499 + end
500 + end
501 +
502 + # The workflow's `concurrency:`, decided as the run is created (GitHub
503 + # decides a workflow's group then, so it can't read `matrix` or `needs`).
504 + # Returns the run, with its group recorded, and what to do about it:
505 + #
506 + # * `waiting` — the reason this run's jobs wait, while another run in
507 + # the group is in progress (nil: they don't)
508 + # * `cancel` — runs to cancel once this one is committed: with
509 + # `cancel-in-progress` every other member, otherwise only one that
510 + # was itself waiting, which this run replaces
511 + # * `note` — why the block was ignored, for the run's jobs to show
512 + defp run_concurrency(%WorkflowRun{} = run, %Repository{} = repo) do
513 + none = %{waiting: nil, cancel: [], cancel_reason: nil, note: nil}
514 +
515 + case Concurrency.evaluate(run.workflow_doc && run.workflow_doc["concurrency"], fn ->
516 + Context.for_run(run, repo)
517 + end) do
518 + :none ->
519 + {run, none}
520 +
521 + {:error, reason} ->
522 + Logger.warning(
523 + "workflow run #{run.id} (#{run.workflow_path}): concurrency ignored: #{reason}"
524 + )
525 +
526 + {run, %{none | note: "Concurrency ignored: " <> reason}}
527 +
528 + {:ok, group, cancel?} ->
529 + run =
530 + run
531 + |> Ecto.Changeset.change(concurrency_group: group, cancel_in_progress: cancel?)
532 + |> Repo.update!()
533 +
534 + Concurrency.lock!(run.repository_id)
535 + members = run_group_members(run.repository_id, group, run.id)
536 + {pending, in_progress} = Enum.split_with(members, fn {_id, pending?} -> pending? end)
537 +
538 + cancel =
539 + if cancel?,
540 + do: Enum.map(members, &elem(&1, 0)),
541 + else: Enum.map(pending, &elem(&1, 0))
542 +
543 + waiting = if not cancel? and in_progress != [], do: "Waiting for " <> group
544 +
545 + {run,
546 + %{
547 + none
548 + | waiting: waiting,
549 + cancel: cancel,
550 + cancel_reason: "Cancelled by run ##{run.number} in #{group}"
551 + }}
552 + end
553 + end
554 +
555 + # The unfinished runs of a group, oldest first, as `{id, pending?}`. A
556 + # pending run is one whose jobs are still held for the group: its jobs
557 + # are `waiting` without a group of their own (a job's own group is only
558 + # decided as it is queued).
559 + defp run_group_members(repository_id, group, except_id) do
560 + from(r in WorkflowRun,
561 + where:
562 + r.repository_id == ^repository_id and r.concurrency_group == ^group and
563 + r.status in ["queued", "running"] and r.id != ^except_id,
564 + order_by: r.id,
565 + select:
566 + {r.id,
567 + fragment(
568 + "EXISTS (SELECT 1 FROM workflow_jobs j WHERE j.workflow_run_id = ? AND j.status = 'waiting' AND j.concurrency_group IS NULL)",
569 + r.id
570 + )}
571 + )
572 + |> Repo.all()
573 + end
574 +
575 + # A run that has finished, or was cancelled, lets the next one in its
576 + # group go: once no member is in progress, the oldest waiting run's jobs
577 + # are queued (each through its own job-level group, decided now).
578 + defp release_run_group(%WorkflowRun{concurrency_group: nil}), do: :ok
579 +
580 + defp release_run_group(%WorkflowRun{concurrency_group: group} = run) do
581 + Repo.transaction(fn ->
582 + Concurrency.lock!(run.repository_id)
583 +
584 + with members <- run_group_members(run.repository_id, group, run.id),
585 + false <- Enum.any?(members, fn {_id, pending?} -> not pending? end),
586 + [{next_id, true} | _] <- members do
587 + from(j in WorkflowJob,
588 + where:
589 + j.workflow_run_id == ^next_id and j.status == "waiting" and
590 + is_nil(j.concurrency_group),
591 + order_by: j.id
592 + )
593 + |> Repo.all()
594 + |> Enum.each(&enqueue_job(&1, status_reason: nil))
595 + end
596 + end)
597 +
598 + :ok
474 599 end
475 600
476 601 # `build (arm64, 28)` — the convention Actions uses, so two legs are
@@ -496,17 +621,24 @@ defmodule GitGud.Workflows do
496 621 # matrix support on this side. Separate rows are what let two legs be
497 622 # claimed by two runners, which is the point when the axis is the
498 623 # machine's architecture.
499 − defp create_job(%WorkflowRun{} = run, %{} = job) do
624 + defp create_job(%WorkflowRun{} = run, %{} = job, hold) do
500 625 for combination <- Matrix.expand(job[:strategy] || job.raw["strategy"]) do
501 − create_job_leg(run, job, combination)
626 + create_job_leg(run, job, combination, hold)
502 627 end
503 628 end
504 629
505 − defp create_job_leg(%WorkflowRun{id: rid}, %{} = job, combination) do
630 + defp create_job_leg(%WorkflowRun{id: rid}, %{} = job, combination, hold) do
506 631 # Jobs with unresolved `needs:` start in `blocked` so claim_next_job
507 632 # can never pick them up before their prerequisites complete.
508 − # Promotion to `queued` happens in `resolve_dependents/1`.
509 − initial_status = if job.needs == [], do: "queued", else: "blocked"
633 + # Promotion to `queued` happens in `resolve_dependents/1`. The others
634 + # are `waiting` while the run's concurrency group holds it back, else
635 + # `queued` (and `create_run/2` then decides their own group).
636 + {initial_status, reason} =
637 + cond do
638 + job.needs != [] -> {"blocked", hold.note}
639 + hold.waiting -> {"waiting", hold.waiting}
640 + true -> {"queued", hold.note}
641 + end
510 642
511 643 # `runs-on` is resolved here rather than by the runner, because it
512 644 # decides *which* runner — there is nobody else to ask yet.
@@ -525,7 +657,7 @@ defmodule GitGud.Workflows do
525 657 matrix: if(combination == %{}, do: nil, else: combination),
526 658 definition: Matrix.pin_definition(job.raw, combination)
527 659 })
528 − |> Ecto.Changeset.change(status: initial_status)
660 + |> Ecto.Changeset.change(status: initial_status, status_reason: reason)
529 661 |> Repo.insert!()
530 662
531 663 job.steps
@@ -1253,7 +1385,7 @@ defmodule GitGud.Workflows do
1253 1385 and timings — and broadcast each change so the workflow LiveView
1254 1386 can light up steps in real time as the runner progresses.
1255 1387 """
1256 − def apply_step_states(%WorkflowJob{id: job_id}, step_states) when is_list(step_states) do
1388 + def apply_step_states(%WorkflowJob{id: job_id} = job, step_states) when is_list(step_states) do
1257 1389 steps =
1258 1390 from(s in WorkflowStep, where: s.workflow_job_id == ^job_id)
1259 1391 |> Repo.all()
@@ -1286,6 +1418,11 @@ defmodule GitGud.Workflows do
1286 1418 end
1287 1419 end)
1288 1420
1421 + # A job the server already finished (cancelled, say) whose runner is
1422 + # still reporting: its steps are its real history, but one it reports
1423 + # as running won't be reported again once the runner stops.
1424 + if WorkflowJob.terminal?(job.status), do: finalize_steps(job)
1425 +
1289 1426 :ok
1290 1427 end
1291 1428
@@ -1328,20 +1465,174 @@ defmodule GitGud.Workflows do
1328 1465
1329 1466 defp timestamp_to_dt(_), do: nil
1330 1467
1468 + @doc """
1469 + Move a job to `status`, and do what follows: a finished job sweeps its
1470 + steps, fail-fasts its matrix, resolves its dependents, lets the next job
1471 + in its concurrency group go, and may finish the run.
1472 +
1473 + The job is re-read under a row lock, so a decision made meanwhile (a
1474 + cancel) is what the transition is checked against, not the caller's
1475 + copy: a finished job doesn't go back (`WorkflowJob.status_changeset/3`),
1476 + and `{:error, changeset}` says it was refused.
1477 +
1478 + `opts`: `:conclusion`; `:reason`, recorded as the job's `status_reason`;
1479 + `fail_fast: false` to leave its matrix siblings alone (a job cancelled by
1480 + its concurrency group is no verdict on the matrix).
1481 + """
1331 1482 def update_job_status(%WorkflowJob{} = job, status, opts \\ []) do
1332 − {:ok, updated} =
1333 − job |> WorkflowJob.status_changeset(status, opts) |> Repo.update()
1483 + {fail_fast?, opts} = Keyword.pop(opts, :fail_fast, true)
1484 + {reason, opts} = Keyword.pop(opts, :reason)
1485 +
1486 + {:ok, result} =
1487 + Repo.transaction(fn ->
1488 + WorkflowJob
1489 + |> where([j], j.id == ^job.id)
1490 + |> lock("FOR UPDATE")
1491 + |> Repo.one!()
1492 + |> WorkflowJob.status_changeset(status, opts)
1493 + |> then(fn cs ->
1494 + if reason, do: Ecto.Changeset.change(cs, status_reason: reason), else: cs
1495 + end)
1496 + |> Repo.update()
1497 + end)
1498 +
1499 + case result do
1500 + {:ok, updated} ->
1501 + broadcast({:job_status, job.workflow_run_id, updated})
1502 +
1503 + if WorkflowJob.terminal?(updated.status) do
1504 + finalize_steps(updated)
1505 + if fail_fast?, do: maybe_fail_fast(updated)
1506 + resolve_dependents(updated)
1507 + release_job_group(updated)
1508 + end
1334 1509
1335 − broadcast({:job_status, job.workflow_run_id, updated})
1510 + maybe_complete_run(updated)
1511 + {:ok, updated}
1336 1512
1337 − if WorkflowJob.terminal?(updated.status) do
1338 − finalize_steps(updated)
1339 − maybe_fail_fast(updated)
1340 − resolve_dependents(updated)
1513 + {:error, _changeset} = refused ->
1514 + refused
1341 1515 end
1516 + end
1517 +
1518 + # ── concurrency: jobs ────────────────────────────────────────────────
1519 +
1520 + # The `status_reason` of a job cancelled by a newer one in its group.
1521 + @superseded "Cancelled by a newer job in "
1522 +
1523 + # Queue a job whose turn has come — no `needs`, or its needs resolved,
1524 + # or its run's group released it — through its own `concurrency:`.
1525 + # `changes` are set with the status (`needs_snapshot:`). In a group
1526 + # (repository-wide, evaluated now so it can read `matrix` and `needs`):
1527 + #
1528 + # * `cancel-in-progress` → the group's other jobs are cancelled, and
1529 + # this one is queued
1530 + # * otherwise → a job already waiting in the group is cancelled (this
1531 + # one replaces it), and this one waits if another is queued or
1532 + # running, else is queued
1533 + #
1534 + # A group that can't be evaluated is ignored, and the job says so.
1535 + defp enqueue_job(%WorkflowJob{} = job, changes \\ []) do
1536 + # What the job will be, for its contexts (`needs` reads the snapshot);
1537 + # `job` itself stays as stored, so the changes are written.
1538 + next = struct(job, changes)
1539 + run = Repo.get!(WorkflowRun, job.workflow_run_id)
1540 +
1541 + contexts = fn ->
1542 + Context.for_job(next, run, Repositories.get_repository!(run.repository_id))
1543 + end
1544 +
1545 + case Concurrency.evaluate(next.definition["concurrency"], contexts) do
1546 + :none when job.status == "queued" and changes == [] ->
1547 + job
1548 +
1549 + :none ->
1550 + put_job(job, changes ++ [status: "queued"])
1342 1551
1343 − maybe_complete_run(updated)
1344 − {:ok, updated}
1552 + {:error, reason} ->
1553 + Logger.warning("workflow job #{job.id} (#{job.name}): concurrency ignored: #{reason}")
1554 +
1555 + put_job(
1556 + job,
1557 + changes ++ [status: "queued", status_reason: "Concurrency ignored: " <> reason]
1558 + )
1559 +
1560 + {:ok, group, cancel?} ->
1561 + {:ok, row} =
1562 + Repo.transaction(fn ->
1563 + Concurrency.lock!(run.repository_id)
1564 + members = job_group_members(run.repository_id, group, job.id)
1565 + wait? = not cancel? and Enum.any?(members, &(&1.status in ["queued", "running"]))
1566 +
1567 + row =
1568 + put_job(
1569 + job,
1570 + changes ++
1571 + [
1572 + status: if(wait?, do: "waiting", else: "queued"),
1573 + concurrency_group: group,
1574 + cancel_in_progress: cancel?,
1575 + status_reason:
1576 + if(wait?, do: "Waiting for " <> group, else: next.status_reason)
1577 + ]
1578 + )
1579 +
1580 + superseded =
1581 + if cancel?, do: members, else: Enum.filter(members, &(&1.status == "waiting"))
1582 +
1583 + for member <- superseded do
1584 + update_job_status(member, "cancelled",
1585 + reason: @superseded <> group,
1586 + fail_fast: false
1587 + )
1588 + end
1589 +
1590 + row
1591 + end)
1592 +
1593 + row
1594 + end
1595 + end
1596 +
1597 + defp put_job(%WorkflowJob{} = job, fields) do
1598 + row = job |> Ecto.Changeset.change(fields) |> Repo.update!()
1599 + broadcast({:job_status, row.workflow_run_id, row})
1600 + row
1601 + end
1602 +
1603 + # The group's unfinished jobs across the repository, oldest first.
1604 + defp job_group_members(repository_id, group, except_id) do
1605 + query =
1606 + from(j in WorkflowJob,
1607 + join: r in WorkflowRun,
1608 + on: r.id == j.workflow_run_id,
1609 + where:
1610 + r.repository_id == ^repository_id and j.concurrency_group == ^group and
1611 + j.status in ["queued", "running", "waiting"],
1612 + order_by: j.id
1613 + )
1614 +
1615 + if except_id, do: query |> where([j], j.id != ^except_id) |> Repo.all(), else: Repo.all(query)
1616 + end
1617 +
1618 + # A finished job in a group lets the oldest waiting one go, once nothing
1619 + # else in the group is queued or running.
1620 + defp release_job_group(%WorkflowJob{concurrency_group: nil}), do: :ok
1621 +
1622 + defp release_job_group(%WorkflowJob{concurrency_group: group} = job) do
1623 + %WorkflowRun{repository_id: repository_id} = Repo.get!(WorkflowRun, job.workflow_run_id)
1624 +
1625 + Repo.transaction(fn ->
1626 + Concurrency.lock!(repository_id)
1627 + members = job_group_members(repository_id, group, job.id)
1628 +
1629 + case Enum.split_with(members, &(&1.status == "waiting")) do
1630 + {[next | _], []} -> put_job(next, status: "queued", status_reason: nil)
1631 + _ -> :ok
1632 + end
1633 + end)
1634 +
1635 + :ok
1345 1636 end
1346 1637
1347 1638 # When a job reaches a terminal state the runner doesn't reliably send
@@ -1384,9 +1675,9 @@ defmodule GitGud.Workflows do
1384 1675 # `strategy.fail-fast` — GitHub's default is to abandon the rest of a
1385 1676 # matrix as soon as one leg fails, because the usual reason to run
1386 1677 # sixteen legs is to find out whether they all pass, and once one
1387 − # hasn't you have your answer. Only siblings that haven't started are
1388 − # cancelled; a leg already running is left to finish, since killing a
1389 − # mid-flight job loses its logs for no gain. A leg that fails under
1678 + # hasn't you have your answer. Legs already running are cancelled too,
1679 + # as on GitHub: their runner hears it on its next `UpdateTask`, stops,
1680 + # and still uploads the log it has. A leg that fails under
1390 1681 # `continue-on-error` doesn't count, as on GitHub: that's what marking a
1391 1682 # leg experimental is for.
1392 1683 defp maybe_fail_fast(%WorkflowJob{status: status} = job)
@@ -1395,7 +1686,7 @@ defmodule GitGud.Workflows do
1395 1686 from(j in WorkflowJob,
1396 1687 where:
1397 1688 j.workflow_run_id == ^job.workflow_run_id and j.job_id == ^job.job_id and
1398 − j.id != ^job.id and j.status in ["queued", "blocked"]
1689 + j.id != ^job.id and j.status in ["queued", "blocked", "waiting", "running"]
1399 1690 )
1400 1691 |> Repo.all()
1401 1692 |> Enum.each(&update_job_status(&1, "cancelled"))
@@ -1415,6 +1706,7 @@ defmodule GitGud.Workflows do
1415 1706
1416 1707 * a need that doesn't exist → `cancelled` (author error)
1417 1708 * every need concluded `success` (`effective_result/1`) → `queued`
1709 + (or `waiting`, by its `concurrency:` group)
1418 1710 * otherwise, the job's `if:` calls `always()`, `failure()` or
1419 1711 `cancelled()` → `queued`, and the runner evaluates the `if:`
1420 1712 against the real `needs` results
@@ -1452,12 +1744,7 @@ defmodule GitGud.Workflows do
1452 1744 {:ok, _} = update_job_status(dep, "skipped")
1453 1745
1454 1746 {:queue, snapshot} ->
1455 − {:ok, _} =
1456 − dep
1457 − |> Ecto.Changeset.change(status: "queued", needs_snapshot: snapshot)
1458 − |> Repo.update()
1459 −
1460 − broadcast({:job_status, rid, Repo.get!(WorkflowJob, dep.id)})
1747 + enqueue_job(dep, needs_snapshot: snapshot)
1461 1748
1462 1749 _blocked_or_decided ->
1463 1750 :ok
@@ -1559,23 +1846,36 @@ defmodule GitGud.Workflows do
1559 1846
1560 1847 # A job failed the run when it failed without `continue-on-error`. A
1561 1848 # job cancelled on its own (fail-fast, a runner that gave up) fails it
1562 − # too, as on GitHub; only a cancelled run concludes `cancelled`.
1849 + # too, as on GitHub; only a cancelled run, or one superseded in a
1850 + # concurrency group, concludes `cancelled`.
1563 1851 # Skipped jobs don't fail anything.
1564 1852 defp maybe_complete_run(%WorkflowJob{workflow_run_id: rid}) do
1565 1853 jobs = from(j in WorkflowJob, where: j.workflow_run_id == ^rid) |> Repo.all()
1566 1854 run = Repo.get!(WorkflowRun, rid)
1567 1855
1568 1856 cond do
1569 − # `blocked` is still pending — its upstream is in flight.
1570 − Enum.any?(jobs, &(&1.status in ~w(queued running blocked))) ->
1857 + # Cancelled is final: a straggler (a runner that hadn't heard yet)
1858 + # doesn't re-settle the run as failed. Re-running a job moves the
1859 + # run back to `running` first.
1860 + run.status == "cancelled" ->
1861 + :ok
1862 +
1863 + # `blocked` is still pending — its upstream is in flight — and
1864 + # `waiting` is held by a concurrency group.
1865 + Enum.any?(jobs, &(&1.status in ~w(queued running blocked waiting))) ->
1571 1866 update_run_commit_status(run)
1572 1867 :ok
1573 1868
1574 1869 Enum.any?(jobs, &(effective_result(&1) == "failure")) ->
1575 1870 finish_run(run, "failure")
1576 1871
1872 + # A run whose jobs were cancelled only by newer jobs in their
1873 + # concurrency groups was superseded, not failed.
1577 1874 Enum.any?(jobs, &(&1.status == "cancelled")) ->
1578 − finish_run(run, if(run.status == "cancelled", do: "cancelled", else: "failure"))
1875 + superseded? =
1876 + jobs |> Enum.filter(&(&1.status == "cancelled")) |> Enum.all?(&superseded?/1)
1877 +
1878 + finish_run(run, if(superseded?, do: "cancelled", else: "failure"))
1579 1879
1580 1880 true ->
1581 1881 run = finish_run(run, "success")
@@ -1584,9 +1884,13 @@ defmodule GitGud.Workflows do
1584 1884 end
1585 1885 end
1586 1886
1887 + defp superseded?(%WorkflowJob{status_reason: @superseded <> _}), do: true
1888 + defp superseded?(_job), do: false
1889 +
1587 1890 defp finish_run(%WorkflowRun{} = run, status) do
1588 1891 run = run |> WorkflowRun.status_changeset(status) |> Repo.update!()
1589 1892 update_run_commit_status(run)
1893 + release_run_group(run)
1590 1894 broadcast_run_status(run)
1591 1895 end
1592 1896
@@ -1799,26 +2103,28 @@ defmodule GitGud.Workflows do
1799 2103 ]
1800 2104 )
1801 2105
1802 − {status, snapshot} =
1803 − case verdict.(job) do
1804 − {:queue, snapshot} -> {"queued", snapshot}
1805 − _ -> {"blocked", job.needs_snapshot}
1806 − end
1807 −
2106 + # Back to `blocked` — the one way out of a finished state
2107 + # (`force:`) — and from there queued, through its concurrency
2108 + # group, when its needs allow.
1808 2109 {:ok, row} =
1809 2110 job
2111 + |> WorkflowJob.status_changeset("blocked", force: true)
1810 2112 |> Ecto.Changeset.change(
1811 − status: status,
1812 − needs_snapshot: snapshot,
1813 2113 conclusion: nil,
1814 2114 started_at: nil,
1815 2115 completed_at: nil,
1816 2116 runner_id: nil,
1817 − attempt: job.attempt + 1
2117 + attempt: job.attempt + 1,
2118 + concurrency_group: nil,
2119 + cancel_in_progress: false,
2120 + status_reason: nil
1818 2121 )
1819 2122 |> Repo.update()
1820 2123
1821 − row
2124 + case verdict.(job) do
2125 + {:queue, snapshot} -> enqueue_job(row, needs_snapshot: snapshot)
2126 + _ -> row
2127 + end
1822 2128 end)
1823 2129
1824 2130 # Flip the run non-terminal so it isn't "complete" and the
@@ -1858,29 +2164,48 @@ defmodule GitGud.Workflows do
1858 2164
1859 2165 @doc """
1860 2166 Cancel a queued or running workflow run. Flips the run + every
1861 − un-finished job + every pending/running step to `cancelled`. Runners
1862 − mid-step on a cancelled job see the status change on their next
1863 − heartbeat and self-terminate — we don't kill containers out-of-band.
2167 + un-finished job + every pending/running step to `cancelled`. A runner
2168 + mid-job hears it in the reply to its next `UpdateTask`
2169 + (`RESULT_CANCELLED`) and stops the job, uploading the log it has — we
2170 + don't kill containers out-of-band.
2171 +
2172 + `opts`: `:reason`, recorded on the jobs it cancels (`status_reason`).
2173 + Lets the next run in its concurrency group go, and the next job in each
2174 + of its jobs' groups.
1864 2175
1865 2176 No-op when the run is already terminal.
1866 2177 """
1867 − def cancel(%WorkflowRun{status: status} = run) when status in ~w(success failure cancelled) do
2178 + def cancel(run, opts \\ [])
2179 +
2180 + def cancel(%WorkflowRun{status: status} = run, _opts)
2181 + when status in ~w(success failure cancelled) do
1868 2182 {:ok, run}
1869 2183 end
1870 2184
1871 − def cancel(%WorkflowRun{} = run) do
2185 + def cancel(%WorkflowRun{} = run, opts) do
2186 + reason = Keyword.get(opts, :reason)
2187 +
1872 2188 Repo.transaction(fn ->
1873 2189 now = DateTime.utc_now(:second)
1874 2190
1875 2191 # Blocked jobs are cancelled with the rest, even one whose `if:`
1876 2192 # says `always()` or `cancelled()`. GitHub would still run those
1877 2193 # once their needs finish; here a cancelled run runs nothing more.
1878 − Repo.update_all(
1879 − from(j in WorkflowJob,
1880 − where: j.workflow_run_id == ^run.id and j.status in ["queued", "running", "blocked"]
1881 − ),
1882 − set: [status: "cancelled", conclusion: "cancelled", completed_at: now]
1883 − )
2194 + {_count, groups} =
2195 + Repo.update_all(
2196 + from(j in WorkflowJob,
2197 + where:
2198 + j.workflow_run_id == ^run.id and
2199 + j.status in ["queued", "running", "blocked", "waiting"],
2200 + select: j.concurrency_group
2201 + ),
2202 + set: [
2203 + status: "cancelled",
2204 + conclusion: "cancelled",
2205 + completed_at: now,
2206 + status_reason: reason
2207 + ]
2208 + )
1884 2209
1885 2210 Repo.update_all(
1886 2211 from(s in WorkflowStep,
@@ -1902,11 +2227,16 @@ defmodule GitGud.Workflows do
1902 2227 {:workflow_job_status, %WorkflowJob{workflow_run_id: run.id}}
1903 2228 )
1904 2229
1905 − updated
2230 + {updated, groups |> Enum.reject(&is_nil/1) |> Enum.uniq()}
1906 2231 end)
1907 2232 |> case do
1908 − {:ok, updated} ->
2233 + {:ok, {updated, groups}} ->
1909 2234 update_run_commit_status(updated)
2235 + release_run_group(updated)
2236 +
2237 + for group <- groups,
2238 + do: release_job_group(%WorkflowJob{workflow_run_id: run.id, concurrency_group: group})
2239 +
1910 2240 notify_run_webhook(updated)
1911 2241 {:ok, updated}
1912 2242
added lib/git_gud/workflows/concurrency.ex
+92 −0

Click to load diff…

added lib/git_gud/workflows/context.ex
+205 −0

Click to load diff…

added lib/git_gud/workflows/expr.ex
+390 −0

Click to load diff…

modified lib/git_gud/workflows/runner_payload.ex
+5 −1

Click to load diff…

modified lib/git_gud/workflows/workflow_job.ex
+27 −1

Click to load diff…

modified lib/git_gud/workflows/workflow_run.ex
+5 −0

Click to load diff…

modified lib/git_gud_web/controllers/runner_controller.ex
+39 −135

Click to load diff…

modified lib/git_gud_web/live/ci_components.ex
+1 −0

Click to load diff…

modified lib/git_gud_web/live/workflow_live/index.ex
+1 −0

Click to load diff…

modified lib/git_gud_web/live/workflow_live/show.ex
+12 −2

Click to load diff…

added priv/repo/migrations/20261017155732_add_concurrency_to_workflow_runs_and_jobs.exs
+25 −0

Click to load diff…

added test/git_gud/workflows/concurrency_test.exs
+319 −0

Click to load diff…

added test/git_gud/workflows/expr_test.exs
+116 −0

Click to load diff…

modified test/git_gud/workflows/matrix_dispatch_test.exs
+4 −3

Click to load diff…

modified test/git_gud/workflows/rerun_cancel_test.exs
+29 −0

Click to load diff…

modified test/git_gud/workflows/runner_payload_test.exs
+2 −0

Click to load diff…

modified test/git_gud_web/controllers/runner_controller_test.exs
+138 −0

Click to load diff…

modified test/git_gud_web/live/workflow_run_anonymous_test.exs
+13 −0

Click to load diff…

Parents: 15e553d