neiam
/gitgud
Git Gud
public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages
⭐
1
Log in to mark this repository.
Keep two jobs on one runner from writing into each other's logs
56749c0 · Gabriel Morell · 2026-09-19 03:15
Message
{commit_body(@commit)}
Files changed
modified
lib/git_gud/workflows.ex
+17
−0
@@ -784,6 +784,23 @@ defmodule GitGud.Workflows do
| 784 | 784 | |
| 785 | 785 | def current_job_for_runner(_), do: nil |
| 786 | 786 | |
| 787 | + @doc """ | |
| 788 | + Every job a runner currently has in flight, newest claim first. | |
| 789 | + | |
| 790 | + A runner with capacity above one holds several at once, which is why | |
| 791 | + this exists alongside `current_job_for_runner/1` — "the" current job | |
| 792 | + is only meaningful when there is one. | |
| 793 | + """ | |
| 794 | + def running_jobs_for_runner(%Runner{id: rid}) do | |
| 795 | + Repo.all( | |
| 796 | + from j in WorkflowJob, | |
| 797 | + where: j.runner_id == ^rid and j.status == "running", | |
| 798 | + order_by: [desc: j.started_at] | |
| 799 | + ) | |
| 800 | + end | |
| 801 | + | |
| 802 | + def running_jobs_for_runner(_), do: [] | |
| 803 | + | |
| 787 | 804 | @doc """ |
| 788 | 805 | Pick the canonical WorkflowStep for log streaming for the given job. |
| 789 | 806 |
modified
lib/git_gud/workflows/workflow_job.ex
+8
−1
@@ -61,9 +61,16 @@ defmodule GitGud.Workflows.WorkflowJob do
| 61 | 61 | |
| 62 | 62 | fields = |
| 63 | 63 | cond do |
| 64 | − status == "running" -> | |
| 64 | + # Only the first `running` report starts the clock. The runner | |
| 65 | + # reports progress for the life of the job, and taking `now` on | |
| 66 | + # each one walked `started_at` forward until a ten-minute job | |
| 67 | + # showed a nine-second duration. | |
| 68 | + status == "running" and is_nil(job.started_at) -> | |
| 65 | 69 | Map.put(fields, :started_at, now) |
| 66 | 70 | |
| 71 | + status == "running" -> | |
| 72 | + fields | |
| 73 | + | |
| 67 | 74 | status in ~w(success failure cancelled skipped) -> |
| 68 | 75 | fields |
| 69 | 76 | |> Map.put(:completed_at, now) |
modified
lib/git_gud_web/controllers/runner_controller.ex
+61
−10
@@ -468,16 +468,7 @@ defmodule GitGudWeb.RunnerController do
| 468 | 468 | with {:ok, %UpdateLogRequest{task_id: tid, index: index, rows: rows}} <- |
| 469 | 469 | decode_request(conn, UpdateLogRequest), |
| 470 | 470 | %RunnerSchema{} = runner <- authenticate(conn) do |
| 471 | − # Forgejo-runner's `task_id` in UpdateLogRequest is *not* the | |
| 472 | − # WorkflowJob id we sent in FetchTaskResponse — it's a runner-local | |
| 473 | − # counter (typically 0 for the only-running job). So we ignore | |
| 474 | − # `tid` for the lookup and use the runner row to find which | |
| 475 | − # WorkflowJob is `running` for this authenticated runner. | |
| 476 | − job = | |
| 477 | − case Workflows.current_job_for_runner(runner) do | |
| 478 | − %GitGud.Workflows.WorkflowJob{} = j -> j | |
| 479 | − _ -> nil | |
| 480 | − end | |
| 471 | + job = log_job(runner, tid) | |
| 481 | 472 | |
| 482 | 473 | step = job && Workflows.ensure_log_step(job) |
| 483 | 474 |
@@ -516,6 +507,66 @@ defmodule GitGudWeb.RunnerController do
| 516 | 507 | end |
| 517 | 508 | end |
| 518 | 509 | |
| 510 | + # Which job these rows belong to. | |
| 511 | + # | |
| 512 | + # `task_id` is the id we put in FetchTaskResponse — a WorkflowJob id — | |
| 513 | + # and the same value `update_task` already trusts as `state.id`. An | |
| 514 | + # earlier reading of it as "a runner-local counter, typically 0" came | |
| 515 | + # from looking the id up in `workflow_steps`, where it was never going | |
| 516 | + # to be found; the consequence was that every row was attributed to | |
| 517 | + # whichever job this runner had started most recently. One job at a | |
| 518 | + # time that is right by accident. Two jobs at once and their streams | |
| 519 | + # merge — and because both number their rows from zero into one | |
| 520 | + # step's `seq`, the loser's rows are dropped by the conflict clause | |
| 521 | + # rather than stored. | |
| 522 | + # | |
| 523 | + # The id is checked against the authenticated runner: it arrives as a | |
| 524 | + # bare integer, and nothing else would stop one runner appending to | |
| 525 | + # another repository's job. | |
| 526 | + defp log_job(runner, tid) when is_integer(tid) and tid > 0 do | |
| 527 | + case GitGud.Repo.get(GitGud.Workflows.WorkflowJob, tid) do | |
| 528 | + %GitGud.Workflows.WorkflowJob{runner_id: rid} = job when rid == runner.id -> | |
| 529 | + job | |
| 530 | + | |
| 531 | + %GitGud.Workflows.WorkflowJob{} -> | |
| 532 | + require Logger | |
| 533 | + | |
| 534 | + Logger.warning( | |
| 535 | + "update_log: job #{tid} is not claimed by runner_id=#{runner.id}; ignoring the id" | |
| 536 | + ) | |
| 537 | + | |
| 538 | + claimed_job(runner) | |
| 539 | + | |
| 540 | + nil -> | |
| 541 | + claimed_job(runner) | |
| 542 | + end | |
| 543 | + end | |
| 544 | + | |
| 545 | + defp log_job(runner, _tid), do: claimed_job(runner) | |
| 546 | + | |
| 547 | + # Fallback for a runner that sends no usable id. With one job claimed | |
| 548 | + # this is exact; with several there is nothing to route by, so it says | |
| 549 | + # so and keeps the old behaviour rather than dropping the rows. | |
| 550 | + defp claimed_job(runner) do | |
| 551 | + case Workflows.running_jobs_for_runner(runner) do | |
| 552 | + [job] -> | |
| 553 | + job | |
| 554 | + | |
| 555 | + [job | _] = jobs -> | |
| 556 | + require Logger | |
| 557 | + | |
| 558 | + Logger.warning( | |
| 559 | + "update_log: #{length(jobs)} jobs running on runner_id=#{runner.id} and no task id " <> | |
| 560 | + "to route by; attributing to job #{job.id}" | |
| 561 | + ) | |
| 562 | + | |
| 563 | + job | |
| 564 | + | |
| 565 | + [] -> | |
| 566 | + nil | |
| 567 | + end | |
| 568 | + end | |
| 569 | + | |
| 519 | 570 | # ── codec ──────────────────────────────────────────────────────────── |
| 520 | 571 | |
| 521 | 572 | # Recognized protobuf content types. Newer Forgejo runners (Connect-Go) |
modified
test/git_gud_web/controllers/runner_controller_test.exs
+78
−2
@@ -187,8 +187,8 @@ defmodule GitGudWeb.RunnerControllerTest do
| 187 | 187 | definition: %{} |
| 188 | 188 | }) |
| 189 | 189 | |> GitGud.Repo.insert!() |
| 190 | − # UpdateLog routes by the runner's current claim, not by an id | |
| 191 | − # in the request, so the test needs the job to look "running". | |
| 190 | + # With no usable task id in the request, UpdateLog falls back to | |
| 191 | + # the runner's single claim, so the job has to look "running". | |
| 192 | 192 | |> Ecto.Changeset.change( |
| 193 | 193 | status: "running", |
| 194 | 194 | runner_id: runner.id, |
@@ -229,5 +229,81 @@ defmodule GitGudWeb.RunnerControllerTest do
| 229 | 229 | |
| 230 | 230 | assert Enum.any?(logs, &(&1 =~ "line 1")) |
| 231 | 231 | end |
| 232 | + | |
| 233 | + test "UpdateLog keeps two jobs on one runner in separate streams", %{ | |
| 234 | + conn: conn, | |
| 235 | + runner_token: token | |
| 236 | + } do | |
| 237 | + # The failure this covers: a runner with capacity for two sends | |
| 238 | + # both jobs' rows down one connection. Routed by "the runner's | |
| 239 | + # current job" they merge into one stream, and since both number | |
| 240 | + # their rows from zero, the second job's rows collide with the | |
| 241 | + # first's `seq` and are dropped rather than stored. | |
| 242 | + user = GitGud.AccountsFixtures.user_fixture() | |
| 243 | + {:ok, repo} = GitGud.Repositories.create_repository(user, %{"name" => "parallel"}) | |
| 244 | + runner = GitGud.Repo.one!(GitGud.Workflows.Runner) | |
| 245 | + | |
| 246 | + run = | |
| 247 | + %GitGud.Workflows.WorkflowRun{repository_id: repo.id} | |
| 248 | + |> GitGud.Workflows.WorkflowRun.create_changeset(%{ | |
| 249 | + number: 1, | |
| 250 | + workflow_path: ".github/workflows/ci.yml", | |
| 251 | + trigger: "push", | |
| 252 | + head_sha: :crypto.strong_rand_bytes(20) | |
| 253 | + }) | |
| 254 | + |> GitGud.Repo.insert!() | |
| 255 | + | |
| 256 | + claim = fn job_id -> | |
| 257 | + job = | |
| 258 | + %GitGud.Workflows.WorkflowJob{workflow_run_id: run.id} | |
| 259 | + |> GitGud.Workflows.WorkflowJob.create_changeset(%{job_id: job_id, definition: %{}}) | |
| 260 | + |> GitGud.Repo.insert!() | |
| 261 | + |> Ecto.Changeset.change( | |
| 262 | + status: "running", | |
| 263 | + runner_id: runner.id, | |
| 264 | + started_at: DateTime.utc_now(:second) | |
| 265 | + ) | |
| 266 | + |> GitGud.Repo.update!() | |
| 267 | + | |
| 268 | + step = | |
| 269 | + %GitGud.Workflows.WorkflowStep{workflow_job_id: job.id} | |
| 270 | + |> GitGud.Workflows.WorkflowStep.create_changeset(%{number: 0, name: job_id}) | |
| 271 | + |> GitGud.Repo.insert!() | |
| 272 | + | |
| 273 | + {job, step} | |
| 274 | + end | |
| 275 | + | |
| 276 | + {first, first_step} = claim.("android") | |
| 277 | + {second, second_step} = claim.("cli") | |
| 278 | + | |
| 279 | + send_rows = fn task_id, content -> | |
| 280 | + body = | |
| 281 | + %UpdateLogRequest{task_id: task_id, index: 0, rows: [%LogRow{content: content}]} | |
| 282 | + |> UpdateLogRequest.encode() | |
| 283 | + |> IO.iodata_to_binary() | |
| 284 | + | |
| 285 | + resp = | |
| 286 | + conn | |
| 287 | + |> put_req_header("content-type", "application/protobuf") | |
| 288 | + |> put_req_header("authorization", "Bearer " <> token) | |
| 289 | + |> post("/api/actions/_apis/runner.v1.RunnerService/UpdateLog", body) | |
| 290 | + | |
| 291 | + assert resp.status == 200 | |
| 292 | + end | |
| 293 | + | |
| 294 | + send_rows.(first.id, "installing the android sdk") | |
| 295 | + send_rows.(second.id, "compiling the cli") | |
| 296 | + | |
| 297 | + bodies = fn step -> | |
| 298 | + GitGud.Repo.all( | |
| 299 | + from l in "workflow_step_logs", | |
| 300 | + where: l.workflow_step_id == ^step.id, | |
| 301 | + select: l.body | |
| 302 | + ) | |
| 303 | + end | |
| 304 | + | |
| 305 | + assert bodies.(first_step) == ["installing the android sdk"] | |
| 306 | + assert bodies.(second_step) == ["compiling the cli"] | |
| 307 | + end | |
| 232 | 308 | end |
| 233 | 309 | end |
Parents: 369e3b3