neiam /gitgud

Git Gud

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

Actions: the stale-job reaper goes by the runner's last report

abee805 · Gabriel Morell · 2026-10-19 15:20

✓ build / android (push) ✓ build / build (push) ✓ build / cli (push) ✓ build / deb (push) ✓ hello / smoke (push)

18 files +160 -4

Message

{commit_body(@commit)}

Files changed

modified lib/git_gud/workflows.ex

+23 −0

@@ -1850,6 +1850,29 @@ defmodule GitGud.Workflows do

1850	1850
1851	1851	def running_jobs_for_runner(_), do: []
1852	1852
	1853	+ @heartbeat_every 30
	1854	+
	1855	+ @doc """
	1856	+ Note that `job`'s runner just reported on it (`UpdateTask`,
	1857	+ `UpdateLog`), so the stale-job reaper leaves it alone. At most one
	1858	+ write per #{@heartbeat_every}s: a runner reports every second or so.
	1859	+ Only a `running` job is touched.
	1860	+ """
	1861	+ def heartbeat(%WorkflowJob{id: id}) do
	1862	+ now = DateTime.utc_now(:second)
	1863	+ since = DateTime.add(now, -@heartbeat_every, :second)
	1864	+
	1865	+ {_count, _} =
	1866	+ from(j in WorkflowJob,
	1867	+ where:
	1868	+ j.id == ^id and j.status == "running" and
	1869	+ (is_nil(j.heartbeat_at) or j.heartbeat_at < ^since)
	1870	+ )
	1871	+ \|> Repo.update_all(set: [heartbeat_at: now])
	1872	+
	1873	+ :ok
	1874	+ end
	1875	+
1853	1876	@doc """
1854	1877	Pick the canonical WorkflowStep for log streaming for the given job.
1855	1878

modified lib/git_gud/workflows/workers/stale_job_reaper.ex

+11 −3

@@ -1,7 +1,13 @@

1	1	defmodule GitGud.Workflows.Workers.StaleJobReaper do
2	2	@moduledoc """
3	3	Resets `workflow_jobs` rows that have been `running` past the
4		− heartbeat threshold without an `UpdateTask` from the runner.
	4	+ heartbeat threshold without an `UpdateTask` or `UpdateLog` from the
	5	+ runner (`heartbeat_at`, falling back to `started_at` for a job claimed
	6	+ before there was one). It used to go by `started_at` alone, which
	7	+ stopped moving once the first `running` report fixed the job's start:
	8	+ every job longer than the window was re-queued under the runner still
	9	+ running it, which then had its reports ignored and its job token
	10	+ refused (a package upload's 401) as soon as both checked the claim.
5	11
6	12	Why this exists: `Workflows.claim_next_job/1` flips a job to
7	13	`running` inside the transaction before the controller has

@@ -44,14 +50,16 @@ defmodule GitGud.Workflows.Workers.StaleJobReaper do

44	50	from(j in WorkflowJob,
45	51	where:
46	52	j.status == "running" and
47		− (is_nil(j.started_at) or j.started_at < ^cutoff),
	53	+ (is_nil(j.started_at) or
	54	+ fragment("coalesce(?, ?)", j.heartbeat_at, j.started_at) < ^cutoff),
48	55	select: j
49	56	)
50	57	\|> Repo.update_all(
51	58	set: [
52	59	status: "queued",
53	60	runner_id: nil,
54		− started_at: nil
	61	+ started_at: nil,
	62	+ heartbeat_at: nil
55	63	]
56	64	)
57	65

modified lib/git_gud/workflows/workflow_job.ex

+4 −1

@@ -33,6 +33,8 @@ defmodule GitGud.Workflows.WorkflowJob do

33	33	field :status, :string, default: "queued"
34	34	field :conclusion, :string
35	35	field :started_at, :utc_datetime
	36	+ # Last report from the runner working on it; see `Workflows.heartbeat/1`.
	37	+ field :heartbeat_at, :utc_datetime
36	38	field :completed_at, :utc_datetime
37	39	field :definition, :map, default: %{}
38	40	# Populated by `update_task/2` from `UpdateTaskRequest.outputs`.

@@ -120,7 +122,8 @@ defmodule GitGud.Workflows.WorkflowJob do

120	122	def terminal?(_), do: false
121	123
122	124	def claim_changeset(job, runner_id) do
123		− change(job, status: "running", runner_id: runner_id, started_at: DateTime.utc_now(:second))
	125	+ now = DateTime.utc_now(:second)
	126	+ change(job, status: "running", runner_id: runner_id, started_at: now, heartbeat_at: now)
124	127	end
125	128
126	129	def valid_statuses, do: @valid_statuses

modified lib/git_gud_web/controllers/runner_controller.ex

+2 −0

@@ -409,6 +409,7 @@ defmodule GitGudWeb.RunnerController do

409	409	respond(conn, %UpdateTaskResponse{state: state, sent_outputs: []})
410	410
411	411	job ->
	412	+ :ok = Workflows.heartbeat(job)
412	413	mapped = map_result(state && state.result)
413	414
414	415	# Per-step status BEFORE the job-level transition. We get

@@ -541,6 +542,7 @@ defmodule GitGudWeb.RunnerController do

541	542	respond(conn, %UpdateLogResponse{ack_index: ack_index})
542	543
543	544	true ->
	545	+ :ok = Workflows.heartbeat(job)
544	546	:ok = Workflows.append_log(step, index, rows)
545	547	respond(conn, %UpdateLogResponse{ack_index: ack_index})
546	548	end

added priv/repo/migrations/20261019151445_add_heartbeat_at_to_workflow_jobs.exs

+13 −0

@@ -0,0 +1,13 @@

	1	+defmodule GitGud.Repo.Migrations.AddHeartbeatAtToWorkflowJobs do
	2	+ use Ecto.Migration
	3	+
	4	+ # When the job's runner last reported on it (UpdateTask / UpdateLog).
	5	+ # The stale-job reaper goes by this, not by `started_at`, so a job
	6	+ # longer than the reaper's window isn't taken from a runner still
	7	+ # working on it.
	8	+ def change do
	9	+ alter table(:workflow_jobs) do
	10	+ add :heartbeat_at, :utc_datetime
	11	+ end
	12	+ end
	13	+end

added test/git_gud/workflows/stale_job_reaper_test.exs

+107 −0

@@ -0,0 +1,107 @@

	1	+defmodule GitGud.Workflows.StaleJobReaperTest do
	2	+ @moduledoc """
	3	+ The reaper takes back jobs whose runner has gone quiet — by the last
	4	+ report (`heartbeat_at`), not by when the job started, so a job longer
	5	+ than the window keeps its runner, its reports and its job token.
	6	+ """
	7	+ use GitGud.DataCase, async: false
	8	+
	9	+ alias GitGud.Workflows
	10	+ alias GitGud.Workflows.RunnerJwt
	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	+ \|> Repo.insert!()
	29	+
	30	+ {:ok, run: run}
	31	+ end
	32	+
	33	+ defp ago(seconds), do: DateTime.add(DateTime.utc_now(:second), -seconds, :second)
	34	+
	35	+ defp running_job(run, started_at, heartbeat_at) do
	36	+ %WorkflowJob{workflow_run_id: run.id}
	37	+ \|> WorkflowJob.create_changeset(%{
	38	+ job_id: "build-#{System.unique_integer([:positive])}",
	39	+ definition: %{}
	40	+ })
	41	+ \|> Ecto.Changeset.change(
	42	+ status: "running",
	43	+ started_at: started_at,
	44	+ heartbeat_at: heartbeat_at
	45	+ )
	46	+ \|> Repo.insert!()
	47	+ end
	48	+
	49	+ defp reap!, do: :ok = StaleJobReaper.perform(%Oban.Job{})
	50	+
	51	+ test "a long job whose runner still reports keeps running, and its token works", %{run: run} do
	52	+ job = running_job(run, ago(600), ago(10))
	53	+ token = RunnerJwt.mint(job, run)
	54	+
	55	+ reap!()
	56	+
	57	+ assert %WorkflowJob{status: "running"} = Repo.get!(WorkflowJob, job.id)
	58	+ assert {:ok, _claims} = RunnerJwt.verify_active(token)
	59	+ end
	60	+
	61	+ test "a job whose runner went quiet is queued again", %{run: run} do
	62	+ job = running_job(run, ago(900), ago(600))
	63	+
	64	+ reap!()
	65	+
	66	+ assert %WorkflowJob{status: "queued", runner_id: nil, started_at: nil, heartbeat_at: nil} =
	67	+ Repo.get!(WorkflowJob, job.id)
	68	+ end
	69	+
	70	+ test "a job claimed before heartbeats goes by when it started", %{run: run} do
	71	+ old = running_job(run, ago(600), nil)
	72	+ fresh = running_job(run, ago(30), nil)
	73	+
	74	+ reap!()
	75	+
	76	+ assert Repo.get!(WorkflowJob, old.id).status == "queued"
	77	+ assert Repo.get!(WorkflowJob, fresh.id).status == "running"
	78	+ end
	79	+
	80	+ describe "heartbeat/1" do
	81	+ test "moves a stale heartbeat to now", %{run: run} do
	82	+ job = running_job(run, ago(600), ago(120))
	83	+
	84	+ :ok = Workflows.heartbeat(job)
	85	+
	86	+ assert DateTime.diff(DateTime.utc_now(), Repo.get!(WorkflowJob, job.id).heartbeat_at) < 5
	87	+ end
	88	+
	89	+ test "writes at most every 30 seconds", %{run: run} do
	90	+ recent = ago(10)
	91	+ job = running_job(run, ago(600), recent)
	92	+
	93	+ :ok = Workflows.heartbeat(job)
	94	+
	95	+ assert Repo.get!(WorkflowJob, job.id).heartbeat_at == recent
	96	+ end
	97	+
	98	+ test "leaves a job that isn't running alone", %{run: run} do
	99	+ job = running_job(run, ago(600), nil)
	100	+ job \|> Ecto.Changeset.change(status: "failure") \|> Repo.update!()
	101	+
	102	+ :ok = Workflows.heartbeat(job)
	103	+
	104	+ assert Repo.get!(WorkflowJob, job.id).heartbeat_at == nil
	105	+ end
	106	+ end
	107	+end

Parents: 4d7e44f