tweaks
b507e61 · gmorell · 2026-05-22 17:04
Files changed
modified
lib/git_gud/workflows.ex
+51
−19
@@ -304,35 +304,67 @@ defmodule GitGud.Workflows do
| 304 | 304 | from j in WorkflowJob |
| 305 | 305 | end |
| 306 | 306 | |
| 307 | − @doc "Append a log chunk to a step." | |
| 308 | − def append_log(%WorkflowStep{id: step_id}, body) when is_binary(body) do | |
| 309 | − seq = next_log_seq(step_id) | |
| 307 | + @doc """ | |
| 308 | + Append log rows to a step at the given starting sequence number. | |
| 309 | + | |
| 310 | + Forgejo's `UpdateLogRequest` carries an `index` field — the runner's | |
| 311 | + view of where this batch of rows starts in the step's total log. We | |
| 312 | + use that as `seq` for the first row, `index + 1` for the second, etc. | |
| 313 | + Retries from the runner with the same `index` re-send identical | |
| 314 | + rows; the `ON CONFLICT (workflow_step_id, seq) DO NOTHING` makes that | |
| 315 | + no-op (idempotent at-least-once). | |
| 316 | + | |
| 317 | + `rows` is the list of `%LogRow{}` protobufs from the request. | |
| 318 | + """ | |
| 319 | + def append_log(%WorkflowStep{id: step_id}, start_seq, rows) when is_integer(start_seq) and is_list(rows) do | |
| 320 | + now = DateTime.utc_now(:second) | |
| 310 | 321 | |
| 311 | − Repo.insert_all( | |
| 312 | − "workflow_step_logs", | |
| 313 | − [ | |
| 322 | + entries = | |
| 323 | + rows | |
| 324 | + |> Enum.with_index(start_seq) | |
| 325 | + |> Enum.map(fn {row, seq} -> | |
| 314 | 326 | %{ |
| 315 | 327 | workflow_step_id: step_id, |
| 316 | 328 | seq: seq, |
| 317 | − body: body, | |
| 318 | − inserted_at: DateTime.utc_now(:second) | |
| 329 | + body: row.content || "", | |
| 330 | + inserted_at: now | |
| 319 | 331 | } |
| 320 | − ] | |
| 321 | − ) | |
| 332 | + end) | |
| 333 | + | |
| 334 | + if entries != [] do | |
| 335 | + Repo.insert_all( | |
| 336 | + "workflow_step_logs", | |
| 337 | + entries, | |
| 338 | + on_conflict: :nothing, | |
| 339 | + conflict_target: [:workflow_step_id, :seq] | |
| 340 | + ) | |
| 341 | + | |
| 342 | + Enum.each(entries, fn %{seq: seq, body: body} -> | |
| 343 | + broadcast({:log, step_id, seq, body}) | |
| 344 | + end) | |
| 345 | + end | |
| 322 | 346 | |
| 323 | − broadcast({:log, step_id, seq, body}) | |
| 324 | 347 | :ok |
| 325 | 348 | end |
| 326 | 349 | |
| 350 | + # Compatibility shim for callers that still use the old single-body | |
| 351 | + # signature (none in tree at the moment, but keeps the contract | |
| 352 | + # explicit). Computes a fresh sequence number and appends. | |
| 353 | + def append_log(%WorkflowStep{id: step_id} = step, body) when is_binary(body) do | |
| 354 | + seq = next_log_seq(step_id) | |
| 355 | + append_log(step, seq, [%{content: body}]) | |
| 356 | + end | |
| 357 | + | |
| 327 | 358 | defp next_log_seq(step_id) do |
| 328 | − Repo.one( | |
| 329 | − from(l in "workflow_step_logs", | |
| 330 | − where: l.workflow_step_id == ^step_id, | |
| 331 | − select: max(l.seq) | |
| 332 | − ) | |
| 333 | − ) || | |
| 334 | − 0 | |
| 335 | − |> Kernel.+(1) | |
| 359 | + max_seq = | |
| 360 | + Repo.one( | |
| 361 | + from(l in "workflow_step_logs", | |
| 362 | + where: l.workflow_step_id == ^step_id, | |
| 363 | + select: max(l.seq) | |
| 364 | + ) | |
| 365 | + ) || -1 | |
| 366 | + | |
| 367 | + max_seq + 1 | |
| 336 | 368 | end |
| 337 | 369 | |
| 338 | 370 | def update_job_status(%WorkflowJob{} = job, status, opts \\ []) do |
modified
lib/git_gud_web/controllers/runner_controller.ex
+8
−5
@@ -275,17 +275,20 @@ defmodule GitGudWeb.RunnerController do
| 275 | 275 | # ── update_log ─────────────────────────────────────────────────────── |
| 276 | 276 | |
| 277 | 277 | def update_log(conn, _params) do |
| 278 | − with {:ok, %UpdateLogRequest{task_id: tid, rows: rows}} <- | |
| 278 | + with {:ok, %UpdateLogRequest{task_id: tid, index: index, rows: rows}} <- | |
| 279 | 279 | decode_request(conn, UpdateLogRequest), |
| 280 | 280 | %RunnerSchema{} = _runner <- authenticate(conn) do |
| 281 | 281 | case GitGud.Repo.get(GitGud.Workflows.WorkflowStep, tid) do |
| 282 | 282 | nil -> |
| 283 | − error(conn, :not_found, "no such step") | |
| 283 | + # Same ack-and-ignore stance as update_task: avoid the runner | |
| 284 | + # retrying forever for a step we no longer track. | |
| 285 | + require Logger | |
| 286 | + Logger.warning("update_log lookup miss step_id=#{inspect(tid)}; acking") | |
| 287 | + respond(conn, %UpdateLogResponse{ack_index: index + length(rows)}) | |
| 284 | 288 | |
| 285 | 289 | step -> |
| 286 | − concat = rows |> Enum.map(& &1.content) |> Enum.join("\n") | |
| 287 | − :ok = Workflows.append_log(step, concat) | |
| 288 | − respond(conn, %UpdateLogResponse{ack_index: length(rows)}) | |
| 290 | + :ok = Workflows.append_log(step, index, rows) | |
| 291 | + respond(conn, %UpdateLogResponse{ack_index: index + length(rows)}) | |
| 289 | 292 | end |
| 290 | 293 | else |
| 291 | 294 | nil -> error(conn, :unauthenticated, "bad token") |
Parents: e90a316