runner: snapshot needs results at promotion (fix multi-stage job skip)
3aa2fef · gmorell · 2026-07-02 15:50
Message
{commit_body(@commit)}
Files changed
modified
lib/git_gud/workflows.ex
+11
−1
@@ -924,9 +924,19 @@ defmodule GitGud.Workflows do
| 924 | 924 | {:ok, _} = update_job_status(dep, "cancelled") |
| 925 | 925 | |
| 926 | 926 | Enum.all?(needs_states, fn n -> WorkflowJob.satisfies_dependency?(n.status) end) -> |
| 927 | + # Snapshot upstream results/outputs now, while they're terminal. | |
| 928 | + # build_needs/1 reads this instead of re-reading upstream status | |
| 929 | + # at fetch time (which raced and could yield RESULT_UNSPECIFIED, | |
| 930 | + # making act skip the job on `if: success()`). | |
| 931 | + snapshot = | |
| 932 | + Map.new(dep.needs, fn jid -> | |
| 933 | + n = Map.get(by_jid, jid) | |
| 934 | + {jid, %{"status" => n.status, "outputs" => n.outputs || %{}}} | |
| 935 | + end) | |
| 936 | + | |
| 927 | 937 | {:ok, _} = |
| 928 | 938 | dep |
| 929 | − |> Ecto.Changeset.change(status: "queued") | |
| 939 | + |> Ecto.Changeset.change(status: "queued", needs_snapshot: snapshot) | |
| 930 | 940 | |> Repo.update() |
| 931 | 941 | |
| 932 | 942 | broadcast({:job_status, rid, Repo.get!(WorkflowJob, dep.id)}) |
modified
lib/git_gud/workflows/workflow_job.ex
+5
−0
@@ -31,6 +31,11 @@ defmodule GitGud.Workflows.WorkflowJob do
| 31 | 31 | # `${{ needs.<id>.outputs.<k> }}` — surfaced in their Task.needs |
| 32 | 32 | # by `build_task/1`. |
| 33 | 33 | field :outputs, :map, default: %{} |
| 34 | + # Snapshot of upstream `needs:` results/outputs taken when this job is | |
| 35 | + # promoted from `blocked` -> `queued` (upstream then terminal). Read by | |
| 36 | + # `build_needs/1`; avoids a racy re-query of upstream status at fetch. | |
| 37 | + # Shape: %{"<need_job_id>" => %{"status" => "...", "outputs" => %{}}}. | |
| 38 | + field :needs_snapshot, :map, default: %{} | |
| 34 | 39 | |
| 35 | 40 | belongs_to :workflow_run, WorkflowRun |
| 36 | 41 | belongs_to :runner, Runner |
modified
lib/git_gud_web/controllers/runner_controller.ex
+13
−15
@@ -226,27 +226,25 @@ defmodule GitGudWeb.RunnerController do
| 226 | 226 | end |
| 227 | 227 | end |
| 228 | 228 | |
| 229 | − defp build_needs(%GitGud.Workflows.WorkflowJob{needs: []}), do: %{} | |
| 230 | − | |
| 231 | − defp build_needs(%GitGud.Workflows.WorkflowJob{workflow_run_id: rid, needs: needs}) do | |
| 232 | − import Ecto.Query | |
| 233 | − | |
| 234 | − upstream_jobs = | |
| 235 | − GitGud.Repo.all( | |
| 236 | − from j in GitGud.Workflows.WorkflowJob, | |
| 237 | − where: j.workflow_run_id == ^rid and j.job_id in ^needs, | |
| 238 | − select: %{job_id: j.job_id, status: j.status, outputs: j.outputs} | |
| 239 | − ) | |
| 240 | − | |
| 241 | − Map.new(upstream_jobs, fn %{job_id: jid, status: status, outputs: outputs} -> | |
| 229 | + # Built from the snapshot captured when the job was promoted to | |
| 230 | + # `queued` (Workflows.resolve_dependents/1), so the upstream results are | |
| 231 | + # the definitively-terminal values — never re-read at fetch time, which | |
| 232 | + # raced and could yield RESULT_UNSPECIFIED (act then skips the job on | |
| 233 | + # `if: success()`). Snapshot shape: | |
| 234 | + # %{"<need>" => %{"status" => "...", "outputs" => %{}}} | |
| 235 | + defp build_needs(%GitGud.Workflows.WorkflowJob{needs_snapshot: snap}) | |
| 236 | + when is_map(snap) and map_size(snap) > 0 do | |
| 237 | + Map.new(snap, fn {jid, data} -> | |
| 242 | 238 | {jid, |
| 243 | 239 | %Runner.V1.TaskNeed{ |
| 244 | − result: status_to_result_enum(status), | |
| 245 | − outputs: outputs || %{} | |
| 240 | + result: status_to_result_enum(Map.get(data, "status")), | |
| 241 | + outputs: Map.get(data, "outputs") || %{} | |
| 246 | 242 | }} |
| 247 | 243 | end) |
| 248 | 244 | end |
| 249 | 245 | |
| 246 | + defp build_needs(_job), do: %{} | |
| 247 | + | |
| 250 | 248 | defp status_to_result_enum("success"), do: :RESULT_SUCCESS |
| 251 | 249 | defp status_to_result_enum("failure"), do: :RESULT_FAILURE |
| 252 | 250 | defp status_to_result_enum("cancelled"), do: :RESULT_CANCELLED |
added
priv/repo/migrations/20260702160000_add_workflow_job_needs_snapshot.exs
+15
−0
@@ -0,0 +1,15 @@
| 1 | +defmodule GitGud.Repo.Migrations.AddWorkflowJobNeedsSnapshot do | |
| 2 | + use Ecto.Migration | |
| 3 | + | |
| 4 | + # Snapshot of each upstream `needs:` job's result + outputs, captured | |
| 5 | + # when `resolve_dependents/1` promotes this job to `queued` (at which | |
| 6 | + # point the upstream is definitively terminal). Read by `build_needs/1` | |
| 7 | + # instead of re-querying upstream status at fetch time, which was racy | |
| 8 | + # and could send RESULT_UNSPECIFIED — making act skip the job via | |
| 9 | + # `if: success()`. | |
| 10 | + def change do | |
| 11 | + alter table(:workflow_jobs) do | |
| 12 | + add :needs_snapshot, :map, default: %{}, null: false | |
| 13 | + end | |
| 14 | + end | |
| 15 | +end |
Parents: 4b94605