neiam
/gitgud
Git Gud
public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages
⭐
1
Log in to mark this repository.
Actions: a job no runner takes fails after a timeout
520cc58 · gmorell · 2026-11-03 15:42
Message
{commit_body(@commit)}
Files changed
modified
config/config.exs
+7
−0
@@ -74,6 +74,9 @@ config :git_gud, Oban,
| 74 | 74 | # can't-send (network blip, encoder bug, runner crash). Threshold |
| 75 | 75 | # is configurable via :stale_after_minutes; default 5 minutes. |
| 76 | 76 | {"* * * * *", GitGud.Workflows.Workers.StaleJobReaper}, |
| 77 | + # Fail workflow_jobs no runner has taken in :queued_timeout_hours | |
| 78 | + # (a `runs-on` nobody has a runner for). | |
| 79 | + {"*/5 * * * *", GitGud.Workflows.Workers.QueuedJobTimeout}, | |
| 77 | 80 | # Fire the workflows' own `on: schedule` crons that are due. |
| 78 | 81 | {"* * * * *", GitGud.Workflows.Workers.ScheduleTick} |
| 79 | 82 | ]} |
@@ -83,6 +86,10 @@ config :git_gud, Oban,
| 83 | 86 | # tolerant of slow runners. Tune per deployment. |
| 84 | 87 | config :git_gud, GitGud.Workflows.Workers.StaleJobReaper, stale_after_minutes: 5 |
| 85 | 88 | |
| 89 | +# How long a CI job may sit queued with no runner taking it (most often a | |
| 90 | +# `runs-on` no registered runner carries) before it fails. nil = forever. | |
| 91 | +config :git_gud, GitGud.Workflows.Workers.QueuedJobTimeout, queued_timeout_hours: 24 | |
| 92 | + | |
| 86 | 93 | # Audit log retention (GitGud.Audit). nil = keep every entry forever; |
| 87 | 94 | # an integer N prunes entries older than N days, daily. |
| 88 | 95 | config :git_gud, GitGud.Audit, retention_days: nil |
modified
lib/git_gud/workflows.ex
+10
−1
@@ -2141,11 +2141,14 @@ defmodule GitGud.Workflows do
| 2141 | 2141 | |
| 2142 | 2142 | `opts`: `:conclusion`; `:reason`, recorded as the job's `status_reason`; |
| 2143 | 2143 | `fail_fast: false` to leave its matrix siblings alone (a job cancelled by |
| 2144 | − its concurrency group is no verdict on the matrix). | |
| 2144 | + its concurrency group is no verdict on the matrix); `:from`, the statuses | |
| 2145 | + the locked row must still be in, else the move is refused (a runner | |
| 2146 | + claimed the queued job a timeout was about to fail). | |
| 2145 | 2147 | """ |
| 2146 | 2148 | def update_job_status(%WorkflowJob{} = job, status, opts \\ []) do |
| 2147 | 2149 | {fail_fast?, opts} = Keyword.pop(opts, :fail_fast, true) |
| 2148 | 2150 | {reason, opts} = Keyword.pop(opts, :reason) |
| 2151 | + {from, opts} = Keyword.pop(opts, :from) | |
| 2149 | 2152 | |
| 2150 | 2153 | {:ok, result} = |
| 2151 | 2154 | Repo.transaction(fn -> |
@@ -2165,6 +2168,11 @@ defmodule GitGud.Workflows do
| 2165 | 2168 | reason -> Ecto.Changeset.change(cs, status_reason: reason) |
| 2166 | 2169 | end |
| 2167 | 2170 | end) |
| 2171 | + |> then(fn cs -> | |
| 2172 | + if is_nil(from) or locked.status in from, | |
| 2173 | + do: cs, | |
| 2174 | + else: Ecto.Changeset.add_error(cs, :status, "has moved on", from: locked.status) | |
| 2175 | + end) | |
| 2168 | 2176 | |> Repo.update() |
| 2169 | 2177 | end) |
| 2170 | 2178 |
@@ -2943,6 +2951,7 @@ defmodule GitGud.Workflows do
| 2943 | 2951 | "awaiting_approval" -> "Waiting for approval" |
| 2944 | 2952 | "running" -> "Running" |
| 2945 | 2953 | "success" -> "Successful" <> took(job, " in ") |
| 2954 | + "failure" when is_nil(job.started_at) -> job.status_reason || "Failing" | |
| 2946 | 2955 | "failure" -> "Failing" <> took(job, " after ") |
| 2947 | 2956 | "warning" -> warning_description(job) |
| 2948 | 2957 | "cancelled" -> job.status_reason || "Cancelled" |
added
lib/git_gud/workflows/workers/queued_job_timeout.ex
+93
−0
@@ -0,0 +1,93 @@
| 1 | +defmodule GitGud.Workflows.Workers.QueuedJobTimeout do | |
| 2 | + @moduledoc """ | |
| 3 | + Fails `workflow_jobs` rows that have sat `queued` past the timeout with | |
| 4 | + no runner taking them: most often a `runs-on` that no registered runner | |
| 5 | + carries (an architecture nobody has a runner for yet). Left alone, such | |
| 6 | + a job stays queued for good, holds its run `running`, and holds every | |
| 7 | + later run in the workflow's concurrency group behind it. | |
| 8 | + | |
| 9 | + The job fails as though a runner had reported it, with a reason that | |
| 10 | + says why (`Workflows.update_job_status/3`): its dependents, its matrix, | |
| 11 | + its concurrency group and its run settle as for any failure. A runner | |
| 12 | + that claims the job while the timeout is deciding wins — only a job | |
| 13 | + still `queued` under the row lock is failed. | |
| 14 | + | |
| 15 | + "Queued since" is the row's `updated_at`: every way a job becomes | |
| 16 | + `queued` writes the row (insert, `Workflows` promoting it from | |
| 17 | + `blocked`/`waiting`/`awaiting_approval`, `StaleJobReaper` taking it | |
| 18 | + back from a quiet runner), and nothing else writes a queued row. | |
| 19 | + | |
| 20 | + Only `queued` is timed. `blocked` and `waiting` wait on other jobs, | |
| 21 | + which settle (or time out) in turn; `awaiting_approval` waits on a | |
| 22 | + person. | |
| 23 | + | |
| 24 | + Configure with: | |
| 25 | + | |
| 26 | + config :git_gud, GitGud.Workflows.Workers.QueuedJobTimeout, | |
| 27 | + queued_timeout_hours: 24 | |
| 28 | + | |
| 29 | + `nil` turns it off. Runs every five minutes via the Oban cron plugin | |
| 30 | + (`config.exs`). | |
| 31 | + """ | |
| 32 | + | |
| 33 | + use Oban.Worker, queue: :ci, max_attempts: 3 | |
| 34 | + | |
| 35 | + import Ecto.Query | |
| 36 | + | |
| 37 | + require Logger | |
| 38 | + | |
| 39 | + alias GitGud.Repo | |
| 40 | + alias GitGud.Workflows | |
| 41 | + alias GitGud.Workflows.WorkflowJob | |
| 42 | + | |
| 43 | + @default_hours 24 | |
| 44 | + | |
| 45 | + @impl Oban.Worker | |
| 46 | + def perform(%Oban.Job{}) do | |
| 47 | + case timeout_hours() do | |
| 48 | + nil -> :ok | |
| 49 | + hours -> time_out(hours) | |
| 50 | + end | |
| 51 | + end | |
| 52 | + | |
| 53 | + defp timeout_hours do | |
| 54 | + Application.get_env(:git_gud, __MODULE__, []) | |
| 55 | + |> Keyword.get(:queued_timeout_hours, @default_hours) | |
| 56 | + end | |
| 57 | + | |
| 58 | + defp time_out(hours) do | |
| 59 | + cutoff = DateTime.add(DateTime.utc_now(), -hours * 3600, :second) | |
| 60 | + | |
| 61 | + jobs = | |
| 62 | + from(j in WorkflowJob, | |
| 63 | + where: j.status == "queued" and j.updated_at < ^cutoff, | |
| 64 | + order_by: j.id | |
| 65 | + ) | |
| 66 | + |> Repo.all() | |
| 67 | + | |
| 68 | + failed = | |
| 69 | + Enum.filter(jobs, fn job -> | |
| 70 | + match?( | |
| 71 | + {:ok, _}, | |
| 72 | + Workflows.update_job_status(job, "failure", | |
| 73 | + from: ["queued"], | |
| 74 | + reason: reason(job, hours) | |
| 75 | + ) | |
| 76 | + ) | |
| 77 | + end) | |
| 78 | + | |
| 79 | + if failed != [] do | |
| 80 | + Logger.warning( | |
| 81 | + "queued_job_timeout failed #{length(failed)} job(s) queued >= #{hours}h: " <> | |
| 82 | + inspect(Enum.map(failed, & &1.id)) | |
| 83 | + ) | |
| 84 | + end | |
| 85 | + | |
| 86 | + :ok | |
| 87 | + end | |
| 88 | + | |
| 89 | + defp reason(%WorkflowJob{runs_on: %{"labels" => [_ | _] = labels}}, hours), | |
| 90 | + do: "No runner picked this job up in #{hours}h (runs-on: #{Enum.join(labels, ", ")})" | |
| 91 | + | |
| 92 | + defp reason(_job, hours), do: "No runner picked this job up in #{hours}h" | |
| 93 | +end |
modified
lib/git_gud/workflows/workers/stale_job_reaper.ex
+6
−3
@@ -46,6 +46,10 @@ defmodule GitGud.Workflows.Workers.StaleJobReaper do
| 46 | 46 | |
| 47 | 47 | cutoff = DateTime.utc_now() |> DateTime.add(-minutes * 60, :second) |
| 48 | 48 | |
| 49 | + now = DateTime.utc_now(:second) | |
| 50 | + | |
| 51 | + # `updated_at` too: it's when the job was queued, which | |
| 52 | + # `QueuedJobTimeout` times from, and `update_all` doesn't touch it. | |
| 49 | 53 | {count, jobs} = |
| 50 | 54 | from(j in WorkflowJob, |
| 51 | 55 | where: |
@@ -59,7 +63,8 @@ defmodule GitGud.Workflows.Workers.StaleJobReaper do
| 59 | 63 | status: "queued", |
| 60 | 64 | runner_id: nil, |
| 61 | 65 | started_at: nil, |
| 62 | − heartbeat_at: nil | |
| 66 | + heartbeat_at: nil, | |
| 67 | + updated_at: now | |
| 63 | 68 | ] |
| 64 | 69 | ) |
| 65 | 70 |
@@ -97,8 +102,6 @@ defmodule GitGud.Workflows.Workers.StaleJobReaper do
| 97 | 102 | # Sweep expired request-recovery rows. These are cheap on their |
| 98 | 103 | # own but accumulate fast on a busy runner; piggyback on the |
| 99 | 104 | # minute-cron to keep the table bounded. |
| 100 | − now = DateTime.utc_now(:second) | |
| 101 | − | |
| 102 | 105 | {pruned, _} = |
| 103 | 106 | from(r in "runner_request_recovery", where: r.expires_at < ^now) |
| 104 | 107 | |> Repo.delete_all() |
added
test/git_gud/workflows/queued_job_timeout_test.exs
+127
−0
@@ -0,0 +1,127 @@
| 1 | +defmodule GitGud.Workflows.QueuedJobTimeoutTest do | |
| 2 | + @moduledoc """ | |
| 3 | + A job no runner takes — a `runs-on` nobody has a runner for — fails | |
| 4 | + once it has been queued past the timeout, and its run and dependents | |
| 5 | + settle as for any failure. | |
| 6 | + """ | |
| 7 | + use GitGud.DataCase, async: false | |
| 8 | + | |
| 9 | + alias GitGud.Workflows | |
| 10 | + alias GitGud.Workflows.Workers.QueuedJobTimeout | |
| 11 | + alias GitGud.Workflows.Workers.StaleJobReaper | |
| 12 | + alias GitGud.Workflows.WorkflowJob | |
| 13 | + alias GitGud.Workflows.WorkflowRun | |
| 14 | + | |
| 15 | + import GitGud.ForgeFixtures | |
| 16 | + | |
| 17 | + setup do | |
| 18 | + {_user, repo} = repository_fixture() | |
| 19 | + | |
| 20 | + run = | |
| 21 | + %WorkflowRun{repository_id: repo.id} | |
| 22 | + |> WorkflowRun.create_changeset(%{ | |
| 23 | + number: 1, | |
| 24 | + workflow_path: ".github/workflows/ci.yml", | |
| 25 | + trigger: "push", | |
| 26 | + head_sha: :crypto.strong_rand_bytes(20) | |
| 27 | + }) | |
| 28 | + |> Ecto.Changeset.change(status: "queued") | |
| 29 | + |> Repo.insert!() | |
| 30 | + | |
| 31 | + {:ok, run: run} | |
| 32 | + end | |
| 33 | + | |
| 34 | + defp hours_ago(h), do: DateTime.add(DateTime.utc_now(:second), -h * 3600, :second) | |
| 35 | + | |
| 36 | + defp job(run, status, updated_at, attrs \\ %{}) do | |
| 37 | + %WorkflowJob{workflow_run_id: run.id} | |
| 38 | + |> WorkflowJob.create_changeset( | |
| 39 | + Map.merge( | |
| 40 | + %{ | |
| 41 | + job_id: "build-#{System.unique_integer([:positive])}", | |
| 42 | + runs_on: %{"labels" => ["riscv64"]}, | |
| 43 | + definition: %{} | |
| 44 | + }, | |
| 45 | + attrs | |
| 46 | + ) | |
| 47 | + ) | |
| 48 | + |> Ecto.Changeset.change(status: status, inserted_at: updated_at, updated_at: updated_at) | |
| 49 | + |> Repo.insert!() | |
| 50 | + end | |
| 51 | + | |
| 52 | + defp time_out!, do: :ok = QueuedJobTimeout.perform(%Oban.Job{}) | |
| 53 | + | |
| 54 | + test "a job queued past the timeout fails, and says why; so does its run", %{run: run} do | |
| 55 | + stuck = job(run, "queued", hours_ago(25)) | |
| 56 | + | |
| 57 | + time_out!() | |
| 58 | + | |
| 59 | + failed = Repo.get!(WorkflowJob, stuck.id) | |
| 60 | + assert failed.status == "failure" | |
| 61 | + assert failed.completed_at | |
| 62 | + assert failed.status_reason =~ "riscv64" | |
| 63 | + assert Repo.get!(WorkflowRun, run.id).status == "failure" | |
| 64 | + end | |
| 65 | + | |
| 66 | + test "a job queued within the timeout stays queued", %{run: run} do | |
| 67 | + fresh = job(run, "queued", hours_ago(23)) | |
| 68 | + | |
| 69 | + time_out!() | |
| 70 | + | |
| 71 | + assert Repo.get!(WorkflowJob, fresh.id).status == "queued" | |
| 72 | + end | |
| 73 | + | |
| 74 | + test "blocked, waiting, awaiting_approval and running jobs aren't timed", %{run: run} do | |
| 75 | + others = | |
| 76 | + for status <- ~w(blocked waiting awaiting_approval running), | |
| 77 | + do: job(run, status, hours_ago(100)) | |
| 78 | + | |
| 79 | + time_out!() | |
| 80 | + | |
| 81 | + for j <- others, do: assert(Repo.get!(WorkflowJob, j.id).status == j.status) | |
| 82 | + end | |
| 83 | + | |
| 84 | + test "a job that needs the timed-out one is skipped", %{run: run} do | |
| 85 | + stuck = job(run, "queued", hours_ago(25), %{job_id: "build"}) | |
| 86 | + after_it = job(run, "blocked", hours_ago(25), %{job_id: "publish", needs: ["build"]}) | |
| 87 | + | |
| 88 | + time_out!() | |
| 89 | + | |
| 90 | + assert Repo.get!(WorkflowJob, stuck.id).status == "failure" | |
| 91 | + assert Repo.get!(WorkflowJob, after_it.id).status == "skipped" | |
| 92 | + end | |
| 93 | + | |
| 94 | + test "a job a runner claimed meanwhile isn't failed", %{run: run} do | |
| 95 | + stale_copy = job(run, "queued", hours_ago(25)) | |
| 96 | + | |
| 97 | + Repo.get!(WorkflowJob, stale_copy.id) | |
| 98 | + |> Ecto.Changeset.change(status: "running") | |
| 99 | + |> Repo.update!() | |
| 100 | + | |
| 101 | + assert {:error, _} = Workflows.update_job_status(stale_copy, "failure", from: ["queued"]) | |
| 102 | + assert Repo.get!(WorkflowJob, stale_copy.id).status == "running" | |
| 103 | + end | |
| 104 | + | |
| 105 | + test "nil turns the timeout off", %{run: run} do | |
| 106 | + stuck = job(run, "queued", hours_ago(24 * 30)) | |
| 107 | + previous = Application.get_env(:git_gud, QueuedJobTimeout) | |
| 108 | + Application.put_env(:git_gud, QueuedJobTimeout, queued_timeout_hours: nil) | |
| 109 | + on_exit(fn -> Application.put_env(:git_gud, QueuedJobTimeout, previous || []) end) | |
| 110 | + | |
| 111 | + time_out!() | |
| 112 | + | |
| 113 | + assert Repo.get!(WorkflowJob, stuck.id).status == "queued" | |
| 114 | + end | |
| 115 | + | |
| 116 | + test "a job the stale-job reaper queues again starts a fresh clock", %{run: run} do | |
| 117 | + reaped = | |
| 118 | + job(run, "running", hours_ago(25)) | |
| 119 | + |> Ecto.Changeset.change(started_at: hours_ago(25), heartbeat_at: hours_ago(25)) | |
| 120 | + |> Repo.update!() | |
| 121 | + | |
| 122 | + :ok = StaleJobReaper.perform(%Oban.Job{}) | |
| 123 | + time_out!() | |
| 124 | + | |
| 125 | + assert Repo.get!(WorkflowJob, reaped.id).status == "queued" | |
| 126 | + end | |
| 127 | +end |
Parents: 003392d