GITGUD

neiam /gitgud

Git Gud

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

fj matching

5775192 · gmorell · 2026-05-22 17:43

⏳ build ⏳ hello
16 files +840 -44

Files changed

modified lib/git_gud/workflows.ex
+350 −7
@@ -131,6 +131,10 @@ defmodule GitGud.Workflows do
131 131
132 132 for job <- jobs, do: create_job(run, job)
133 133
134 + # Initial pending commit status so the SCM view shows the run
135 + # is in flight before the first claim lands.
136 + update_run_commit_status(run)
137 +
134 138 run
135 139 end)
136 140 end
@@ -144,6 +148,11 @@ defmodule GitGud.Workflows do
144 148 end
145 149
146 150 defp create_job(%WorkflowRun{id: rid}, %{} = job) do
151 + # Jobs with unresolved `needs:` start in `blocked` so claim_next_job
152 + # can never pick them up before their prerequisites complete.
153 + # Promotion to `queued` happens in `resolve_dependents/1`.
154 + initial_status = if job.needs == [], do: "queued", else: "blocked"
155 +
147 156 job_row =
148 157 %WorkflowJob{workflow_run_id: rid}
149 158 |> WorkflowJob.create_changeset(%{
@@ -153,6 +162,7 @@ defmodule GitGud.Workflows do
153 162 needs: job.needs,
154 163 definition: job.raw
155 164 })
165 + |> Ecto.Changeset.change(status: initial_status)
156 166 |> Repo.insert!()
157 167
158 168 job.steps
@@ -238,6 +248,92 @@ defmodule GitGud.Workflows do
238 248
239 249 def claim_next_job(_runner), do: nil
240 250
251 + @doc """
252 + Claim up to `n` jobs in one batch — for runners that advertise
253 + `task_capacity > 1` in their FetchTaskRequest. Returns a (possibly
254 + empty) list, in claim order.
255 +
256 + Each job is claimed in its own transaction (cheap rollback if mid-
257 + loop encoding later fails); they all become `running` on success.
258 + """
259 + def claim_jobs_for(%Runner{} = runner, n) when is_integer(n) and n > 0 do
260 + claim_jobs_for(runner, n, [])
261 + end
262 +
263 + def claim_jobs_for(_runner, _n), do: []
264 +
265 + defp claim_jobs_for(_runner, 0, acc), do: Enum.reverse(acc)
266 +
267 + defp claim_jobs_for(runner, remaining, acc) do
268 + case claim_next_job(runner) do
269 + %WorkflowJob{} = job -> claim_jobs_for(runner, remaining - 1, [job | acc])
270 + _ -> Enum.reverse(acc)
271 + end
272 + end
273 +
274 + @doc """
275 + Claim a specific job identified by `handle` — typically the
276 + workflow_job_id encoded as a string. Used by `FetchSingleTask` for
277 + targeted retrieval. Returns nil if no queued job matches.
278 +
279 + We accept either a numeric string (interpreted as the WorkflowJob
280 + pk) or a `<run_id>:<job_id>` form (yaml-level routing). Falls
281 + back to normal `claim_next_job/1` for empty / unparseable handles.
282 + """
283 + def claim_job_by_handle(%Runner{} = runner, handle)
284 + when is_binary(handle) and handle != "" do
285 + target_id =
286 + case Integer.parse(handle) do
287 + {n, ""} -> n
288 + _ -> nil
289 + end
290 +
291 + cond do
292 + is_integer(target_id) ->
293 + do_claim_specific_job(runner, target_id)
294 +
295 + true ->
296 + # Unsupported handle shape; fall through to a normal claim.
297 + claim_next_job(runner)
298 + end
299 + end
300 +
301 + def claim_job_by_handle(runner, _), do: claim_next_job(runner)
302 +
303 + defp do_claim_specific_job(%Runner{id: rid} = runner, job_id) do
304 + %Runner{labels: %{"labels" => labels}} = runner
305 + match_labels = if "self-hosted" in labels, do: labels, else: ["self-hosted" | labels]
306 +
307 + Repo.transaction(fn ->
308 + job =
309 + scoped_job_query(runner)
310 + |> where([j], j.id == ^job_id)
311 + |> where([j], j.status == "queued")
312 + |> where([j], fragment("(?->>'value') = ANY(?)", j.runs_on, ^match_labels))
313 + |> lock("FOR UPDATE SKIP LOCKED")
314 + |> Repo.one()
315 +
316 + case job do
317 + nil ->
318 + nil
319 +
320 + job ->
321 + {:ok, claimed} = job |> WorkflowJob.claim_changeset(rid) |> Repo.update()
322 + run = Repo.get!(WorkflowRun, claimed.workflow_run_id)
323 +
324 + if run.status == "queued" do
325 + run |> WorkflowRun.status_changeset("running") |> Repo.update!()
326 + end
327 +
328 + claimed |> Repo.preload(:steps)
329 + end
330 + end)
331 + |> case do
332 + {:ok, result} -> result
333 + _ -> nil
334 + end
335 + end
336 +
241 337 defp do_claim_next_job(%Runner{id: rid} = runner, labels) do
242 338 # GitHub-Actions semantics: `runs-on: self-hosted` matches any
243 339 # self-registered runner regardless of OS label. So if the runner
@@ -367,36 +463,283 @@ defmodule GitGud.Workflows do
367 463 max_seq + 1
368 464 end
369 465
466 + @doc """
467 + Look up a previously-claimed job by `(runner_id, request_key)`. If
468 + the runner retries FetchTask with the same idempotency key (because
469 + our response was dropped in flight), we re-return the same job
470 + instead of orphaning the prior claim.
471 +
472 + Returns the WorkflowJob row if a non-expired recovery row exists
473 + AND the job is still `running` and assigned to this runner.
474 + Otherwise nil.
475 + """
476 + def recover_claimed_job(%Runner{id: rid}, key) when is_binary(key) and key != "" do
477 + now = DateTime.utc_now(:second)
478 +
479 + Repo.one(
480 + from rec in "runner_request_recovery",
481 + join: j in WorkflowJob,
482 + on: j.id == rec.workflow_job_id,
483 + where:
484 + rec.runner_id == ^rid and rec.request_key == ^key and rec.expires_at > ^now and
485 + j.runner_id == ^rid and j.status == "running",
486 + select: j.id
487 + )
488 + |> case do
489 + nil -> nil
490 + jid -> Repo.get(WorkflowJob, jid) |> Repo.preload(:steps)
491 + end
492 + end
493 +
494 + def recover_claimed_job(_, _), do: nil
495 +
496 + @doc """
497 + Stamp the (runner_id, request_key) -> job mapping so a retry of the
498 + same FetchTask gets the same job back.
499 +
500 + `ttl_seconds` defaults to 5 minutes — long enough for any
501 + intermittent network blip, short enough that stale rows don't
502 + accumulate.
503 + """
504 + def stamp_request_recovery(runner, key, job, ttl_seconds \\ 300)
505 +
506 + def stamp_request_recovery(%Runner{id: rid}, key, %WorkflowJob{id: jid}, ttl_seconds)
507 + when is_binary(key) and key != "" do
508 + expires_at = DateTime.utc_now(:second) |> DateTime.add(ttl_seconds, :second)
509 +
510 + Repo.insert_all(
511 + "runner_request_recovery",
512 + [
513 + %{
514 + runner_id: rid,
515 + request_key: key,
516 + workflow_job_id: jid,
517 + expires_at: expires_at
518 + }
519 + ],
520 + on_conflict: {:replace, [:workflow_job_id, :expires_at]},
521 + conflict_target: [:runner_id, :request_key]
522 + )
523 +
524 + :ok
525 + end
526 +
527 + def stamp_request_recovery(_runner, _key, _job, _ttl_seconds), do: :ok
528 +
529 + @doc """
530 + The WorkflowJob currently being executed by the given runner.
531 + Returns nil if there's no running claim.
532 + """
533 + def current_job_for_runner(%Runner{id: rid}) do
534 + Repo.one(
535 + from j in WorkflowJob,
536 + where: j.runner_id == ^rid and j.status == "running",
537 + order_by: [desc: j.started_at],
538 + limit: 1
539 + )
540 + end
541 +
542 + def current_job_for_runner(_), do: nil
543 +
544 + @doc """
545 + Pick the canonical WorkflowStep for log streaming for the given job.
546 +
547 + Forgejo-runner's `UpdateLog` doesn't carry per-step routing — all log
548 + rows for a job arrive on one stream. We funnel them into the first
549 + step (`number == 0`) of the job. If the job has no steps yet (e.g.,
550 + the workflow YAML didn't materialize them at parse time), create a
551 + synthetic step so logs land somewhere queryable.
552 + """
553 + def ensure_log_step(%WorkflowJob{id: jid}) do
554 + case Repo.one(
555 + from s in WorkflowStep,
556 + where: s.workflow_job_id == ^jid,
557 + order_by: [asc: s.number],
558 + limit: 1
559 + ) do
560 + %WorkflowStep{} = step ->
561 + step
562 +
563 + nil ->
564 + %WorkflowStep{workflow_job_id: jid}
565 + |> WorkflowStep.create_changeset(%{number: 0, name: "log"})
566 + |> Repo.insert!()
567 + end
568 + end
569 +
570 + @doc """
571 + Merge `new_outputs` (a `%{string => string}` map) into the job's
572 + stored outputs. Upstream forgejo enforces 255-char key / 1MB value
573 + limits; we apply the same so a hostile workflow can't unbounded-grow
574 + the JSONB column.
575 + """
576 + def set_outputs(%WorkflowJob{} = job, new_outputs) when is_map(new_outputs) do
577 + sanitized =
578 + new_outputs
579 + |> Enum.filter(fn
580 + {k, v} when is_binary(k) and is_binary(v) ->
581 + byte_size(k) <= 255 and byte_size(v) <= 1_048_576
582 +
583 + _ ->
584 + false
585 + end)
586 + |> Map.new()
587 +
588 + merged = Map.merge(job.outputs || %{}, sanitized)
589 +
590 + job
591 + |> Ecto.Changeset.change(outputs: merged)
592 + |> Repo.update()
593 + end
594 +
370 595 def update_job_status(%WorkflowJob{} = job, status, opts \\ []) do
371 596 {:ok, updated} =
372 597 job |> WorkflowJob.status_changeset(status, opts) |> Repo.update()
373 598
374 599 broadcast({:job_status, job.workflow_run_id, updated})
600 +
601 + if WorkflowJob.terminal?(updated.status) do
602 + resolve_dependents(updated)
603 + end
604 +
375 605 maybe_complete_run(updated)
376 606 {:ok, updated}
377 607 end
378 608
609 + @doc """
610 + After a job reaches a terminal state, walk every sibling job in the
611 + same run whose `needs:` array contains this job's `job_id` and
612 + promote them from `blocked`:
613 +
614 + * all listed needs are now `success`/`skipped` → `queued`
615 + * any listed need is `failure`/`cancelled` → `cancelled`
616 + (cascading failure, matches GitHub Actions default)
617 + * mixed-but-not-failed (still some pending needs) → leave `blocked`
618 +
619 + Public so tests can drive it directly.
620 + """
621 + def resolve_dependents(%WorkflowJob{workflow_run_id: rid, job_id: completed_jid}) do
622 + # All sibling jobs in this run, indexed by their job_id (yaml id).
623 + siblings =
624 + from(j in WorkflowJob, where: j.workflow_run_id == ^rid)
625 + |> Repo.all()
626 +
627 + by_jid = Map.new(siblings, fn j -> {j.job_id, j} end)
628 +
629 + dependents = Enum.filter(siblings, fn j -> completed_jid in j.needs and j.status == "blocked" end)
630 +
631 + Enum.each(dependents, fn dep ->
632 + needs_states = Enum.map(dep.needs, fn jid -> Map.get(by_jid, jid) end)
633 +
634 + cond do
635 + Enum.any?(needs_states, &is_nil/1) ->
636 + # A referenced need doesn't exist as a sibling — workflow
637 + # author error. Cancel rather than block forever.
638 + {:ok, _} = update_job_status(dep, "cancelled")
639 +
640 + Enum.any?(needs_states, fn n -> n.status in ~w(failure cancelled) end) ->
641 + {:ok, _} = update_job_status(dep, "cancelled")
642 +
643 + Enum.all?(needs_states, fn n -> WorkflowJob.satisfies_dependency?(n.status) end) ->
644 + {:ok, _} =
645 + dep
646 + |> Ecto.Changeset.change(status: "queued")
647 + |> Repo.update()
648 +
649 + broadcast({:job_status, rid, Repo.get!(WorkflowJob, dep.id)})
650 +
651 + true ->
652 + # Some upstream is still pending. Leave blocked.
653 + :ok
654 + end
655 + end)
656 +
657 + :ok
658 + end
659 +
379 660 defp maybe_complete_run(%WorkflowJob{workflow_run_id: rid}) do
380 661 statuses =
381 662 from(j in WorkflowJob, where: j.workflow_run_id == ^rid, select: j.status)
382 663 |> Repo.all()
383 664
384 665 cond do
385 − Enum.any?(statuses, &(&1 in ~w(queued running))) ->
666 + # `blocked` is still pending — its upstream is in flight.
667 + Enum.any?(statuses, &(&1 in ~w(queued running blocked))) ->
668 + run = Repo.get!(WorkflowRun, rid)
669 + update_run_commit_status(run)
386 670 :ok
387 671
388 672 Enum.any?(statuses, &(&1 in ~w(failure cancelled))) ->
389 − Repo.get!(WorkflowRun, rid)
390 − |> WorkflowRun.status_changeset("failure")
391 − |> Repo.update!()
673 + run =
674 + Repo.get!(WorkflowRun, rid)
675 + |> WorkflowRun.status_changeset("failure")
676 + |> Repo.update!()
677 +
678 + update_run_commit_status(run)
392 679
393 680 true ->
394 − Repo.get!(WorkflowRun, rid)
395 − |> WorkflowRun.status_changeset("success")
396 − |> Repo.update!()
681 + run =
682 + Repo.get!(WorkflowRun, rid)
683 + |> WorkflowRun.status_changeset("success")
684 + |> Repo.update!()
685 +
686 + update_run_commit_status(run)
397 687 end
398 688 end
399 689
690 + @doc """
691 + Upsert the commit status row for the given workflow run. Runs on
692 + every status transition (queued → running → success/failure) so the
693 + SCM views see ✓/✗ as soon as the run finishes.
694 +
695 + `state` mapping:
696 + queued, running -> "pending"
697 + success -> "success"
698 + failure, cancelled -> "failure"
699 + other (error) -> "error"
700 + """
701 + def update_run_commit_status(%WorkflowRun{} = run) do
702 + state =
703 + case run.status do
704 + s when s in ~w(queued running) -> "pending"
705 + "success" -> "success"
706 + s when s in ~w(failure cancelled) -> "failure"
707 + _ -> "error"
708 + end
709 +
710 + context = "gitgud-actions/" <> (run.workflow_name || run.workflow_path || "workflow")
711 +
712 + target_url =
713 + GitGudWeb.Endpoint.url() <>
714 + "/r/repo/" <> Integer.to_string(run.repository_id) <>
715 + "/actions/runs/" <> Integer.to_string(run.id)
716 +
717 + description =
718 + case run.status do
719 + "success" -> "All checks passed"
720 + "failure" -> "Some checks failed"
721 + "cancelled" -> "Cancelled"
722 + _ -> "Run #{run.number} #{run.status}"
723 + end
724 +
725 + attrs = %{
726 + repository_id: run.repository_id,
727 + sha: run.head_sha,
728 + context: context,
729 + state: state,
730 + description: description,
731 + target_url: target_url
732 + }
733 +
734 + %GitGud.Workflows.CommitStatus{}
735 + |> GitGud.Workflows.CommitStatus.changeset(attrs)
736 + |> Repo.insert(
737 + on_conflict:
738 + {:replace, [:state, :description, :target_url, :updated_at]},
739 + conflict_target: [:repository_id, :sha, :context]
740 + )
741 + end
742 +
400 743 # ── pubsub ───────────────────────────────────────────────────────────
401 744
402 745 def topic_for_run(run_id), do: "workflow_run:#{run_id}"
added lib/git_gud/workflows/commit_status.ex
+55 −0
@@ -0,0 +1,55 @@
1 +defmodule GitGud.Workflows.CommitStatus do
2 + @moduledoc """
3 + Per-commit CI / build status — one row per `(repository, sha, context)`.
4 +
5 + `state` is the GitHub-Actions-flavored summary:
6 + * `pending` — the run is queued or running
7 + * `success` — all jobs in the run finished with `success`/`skipped`
8 + * `failure` — at least one job ended in `failure`/`cancelled`
9 + * `error` — internal error (server-side; not produced by runs today)
10 +
11 + `context` identifies the provider; we use `gitgud-actions/<workflow>`
12 + per workflow so multiple workflows on the same SHA stack instead of
13 + overwriting one another.
14 +
15 + `target_url` links to the workflow run view in the UI.
16 +
17 + Persisted by `Workflows.update_run_commit_status/1` on every
18 + WorkflowRun status transition.
19 + """
20 +
21 + use Ecto.Schema
22 +
23 + import Ecto.Changeset
24 +
25 + alias GitGud.Repositories.Repository
26 +
27 + @valid_states ~w(pending success failure error)
28 +
29 + schema "commit_statuses" do
30 + field :sha, :binary
31 + field :context, :string
32 + field :state, :string
33 + field :description, :string
34 + field :target_url, :string
35 +
36 + belongs_to :repository, Repository
37 +
38 + timestamps(type: :utc_datetime)
39 + end
40 +
41 + def changeset(status, attrs) do
42 + status
43 + |> cast(attrs, [:sha, :context, :state, :description, :target_url, :repository_id])
44 + |> validate_required([:sha, :context, :state, :repository_id])
45 + |> validate_inclusion(:state, @valid_states)
46 + |> validate_length(:context, max: 255)
47 + |> validate_length(:description, max: 512)
48 + |> validate_length(:target_url, max: 1024)
49 + |> unique_constraint([:repository_id, :sha, :context],
50 + name: :commit_statuses_repository_id_sha_context_index
51 + )
52 + end
53 +
54 + def valid_states, do: @valid_states
55 +end
modified lib/git_gud/workflows/runner.ex
+17 −1
@@ -12,7 +12,14 @@ defmodule GitGud.Workflows.Runner do
12 12 field :uuid, Ecto.UUID
13 13 field :name, :string
14 14 field :version, :string
15 + # Operator-assigned. Source of truth for `claim_next_job/1`
16 + # routing. Set via the offline-mint UI / mix task and never
17 + # touched by the runner's `Declare` RPC.
15 18 field :labels, :map, default: %{"labels" => ["ubuntu-latest"]}
19 + # Runner-self-reported. The runner re-asserts these every
20 + # `Declare`; we record but don't route on them. Surfaced in
21 + # the UI for operator visibility.
22 + field :agent_labels, :map, default: %{"labels" => []}
16 23 field :token_hash, :binary
17 24 field :last_seen_at, :utc_datetime
18 25 field :status, :string, default: "active"
@@ -26,7 +33,16 @@ defmodule GitGud.Workflows.Runner do
26 33
27 34 def changeset(runner, attrs) do
28 35 runner
29 − |> cast(attrs, [:name, :version, :labels, :status, :repository_id, :organization_id, :uuid])
36 + |> cast(attrs, [
37 + :name,
38 + :version,
39 + :labels,
40 + :agent_labels,
41 + :status,
42 + :repository_id,
43 + :organization_id,
44 + :uuid
45 + ])
30 46 |> validate_required([:name])
31 47 |> validate_inclusion(:status, @valid_statuses)
32 48 |> validate_length(:name, max: 100)
modified lib/git_gud/workflows/workers/stale_job_reaper.ex
+14 −0
@@ -78,6 +78,20 @@ defmodule GitGud.Workflows.Workers.StaleJobReaper do
78 78 |> Repo.update_all(set: [status: "queued"])
79 79 end
80 80
81 + # Sweep expired request-recovery rows. These are cheap on their
82 + # own but accumulate fast on a busy runner; piggyback on the
83 + # minute-cron to keep the table bounded.
84 + now = DateTime.utc_now(:second)
85 +
86 + {pruned, _} =
87 + from(r in "runner_request_recovery", where: r.expires_at < ^now)
88 + |> Repo.delete_all()
89 +
90 + if pruned > 0 do
91 + require Logger
92 + Logger.debug("stale_job_reaper pruned #{pruned} expired request-recovery rows")
93 + end
94 +
81 95 :ok
82 96 end
83 97 end
modified lib/git_gud/workflows/workflow_job.ex
+28 −2
@@ -6,7 +6,15 @@ defmodule GitGud.Workflows.WorkflowJob do
6 6 alias GitGud.Workflows.WorkflowRun
7 7 alias GitGud.Workflows.WorkflowStep
8 8
9 − @valid_statuses ~w(queued running success failure cancelled)
9 + # `blocked` = waiting on upstream `needs:`. Promoted to `queued` (or
10 + # `cancelled`) by `Workflows.resolve_dependents/1` when any sibling
11 + # job completes. `claim_next_job/1` only picks `queued`, so blocked
12 + # jobs are never accidentally claimed by a runner before their
13 + # prerequisites finish.
14 + # `skipped` = upstream that the GitHub-Actions `if:` evaluated false
15 + # for. Treated as success for the purpose of resolving downstream
16 + # `needs:` (matches GitHub semantics).
17 + @valid_statuses ~w(queued running success failure cancelled blocked skipped)
10 18
11 19 schema "workflow_jobs" do
12 20 field :job_id, :string
@@ -18,6 +26,11 @@ defmodule GitGud.Workflows.WorkflowJob do
18 26 field :started_at, :utc_datetime
19 27 field :completed_at, :utc_datetime
20 28 field :definition, :map, default: %{}
29 + # Populated by `update_task/2` from `UpdateTaskRequest.outputs`.
30 + # Downstream jobs in the same run read this via
31 + # `${{ needs.<id>.outputs.<k> }}` — surfaced in their Task.needs
32 + # by `build_task/1`.
33 + field :outputs, :map, default: %{}
21 34
22 35 belongs_to :workflow_run, WorkflowRun
23 36 belongs_to :runner, Runner
@@ -41,7 +54,7 @@ defmodule GitGud.Workflows.WorkflowJob do
41 54 status == "running" ->
42 55 Map.put(fields, :started_at, now)
43 56
44 − status in ~w(success failure cancelled) ->
57 + status in ~w(success failure cancelled skipped) ->
45 58 fields
46 59 |> Map.put(:completed_at, now)
47 60 |> Map.put(:conclusion, Keyword.get(opts, :conclusion, status))
@@ -53,6 +66,19 @@ defmodule GitGud.Workflows.WorkflowJob do
53 66 change(job, fields)
54 67 end
55 68
69 + @doc "Terminal statuses — job won't transition out of these."
70 + def terminal?(s) when s in ~w(success failure cancelled skipped), do: true
71 + def terminal?(_), do: false
72 +
73 + @doc """
74 + Did this job's terminal state count as 'satisfying' a downstream
75 + `needs:` declaration? Mirrors GitHub Actions semantics: a need is
76 + satisfied by `success` or `skipped`, never by `failure` or
77 + `cancelled` (the downstream gets cascade-cancelled instead).
78 + """
79 + def satisfies_dependency?(s) when s in ~w(success skipped), do: true
80 + def satisfies_dependency?(_), do: false
81 +
56 82 def claim_changeset(job, runner_id) do
57 83 change(job, status: "running", runner_id: runner_id, started_at: DateTime.utc_now(:second))
58 84 end
modified lib/git_gud_web/controllers/runner_controller.ex
+163 −30

Click to load diff…

modified lib/git_gud_web/live/org_live/settings_runners.ex
+11 −1

Click to load diff…

modified lib/git_gud_web/live/repo_live/settings_runners.ex
+11 −1

Click to load diff…

modified lib/git_gud_web/protos/runner/v1/messages.pb.ex
+26 −0

Click to load diff…

modified lib/git_gud_web/router.ex
+2 −0

Click to load diff…

modified priv/proto/runner/v1/messages.proto
+21 −0

Click to load diff…

added priv/repo/migrations/20260522170000_add_runner_agent_labels.exs
+38 −0

Click to load diff…

added priv/repo/migrations/20260522180000_add_workflow_job_outputs.exs
+22 −0

Click to load diff…

added priv/repo/migrations/20260522190000_add_runner_request_recovery.exs
+30 −0

Click to load diff…

added priv/repo/migrations/20260522200000_create_commit_statuses.exs
+40 −0

Click to load diff…

modified test/git_gud_web/controllers/runner_controller_test.exs
+12 −2

Click to load diff…

Parents: b507e61