GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Let a job stuck in "running" be cleared

5f05b9a · Gabriel Morell · 2026-09-01 20:45

12 files +171 -3
Message 
{commit_body(@commit)}

Files changed

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/work.ex
+62 −3
@@ -80,6 +80,10 @@ defmodule RoomSanctumWeb.SourceLive.Work do
80 80 end
81 81 end
82 82
83 +
84 + @impl true
85 + def handle_info(_msg, socket), do: {:noreply, socket}
86 +
83 87 defp mine?(socket, id) do
84 88 case Integer.parse(to_string(id)) do
85 89 {id, ""} -> Map.has_key?(socket.assigns.names, id)
@@ -87,9 +91,6 @@ defmodule RoomSanctumWeb.SourceLive.Work do
87 91 end
88 92 end
89 93
90 − @impl true
91 − def handle_info(_msg, socket), do: {:noreply, socket}
92 −
93 94 @impl true
94 95 def handle_event("requeue", %{"id" => id}, socket) do
95 96 case RoomGtfs.Worker.update_static_data(String.to_integer(id)) do
@@ -104,6 +105,64 @@ defmodule RoomSanctumWeb.SourceLive.Work do
104 105 end
105 106 end
106 107
108 + @doc """
109 + Give up on a job that is running and is not.
110 +
111 + Oban has no Lifeline plugin configured here, so nothing rescues a job whose
112 + process died -- a VM crash mid-import leaves the row in `executing` for good.
113 + That is not merely untidy: ImportJob's uniqueness covers `executing` with no
114 + expiry, so the stuck row blocks every future import of that feed, and the
115 + requeue button reports a conflict rather than doing anything.
116 +
117 + Cancelling moves it to a state uniqueness does not count, which is what frees
118 + the feed. It does not stop a genuinely running import -- if the process is
119 + alive it is asked to stop, which for a bulk load in progress means the next
120 + import starts from a truncate anyway.
121 + """
122 + @impl true
123 + def handle_event("clear", %{"id" => id}, socket) do
124 + case Oban.cancel_job(String.to_integer(id)) do
125 + :ok ->
126 + {:noreply,
127 + socket
128 + |> put_flash(:info, "Cleared. That feed can be queued again.")
129 + |> load()}
130 +
131 + _otherwise ->
132 + {:noreply, put_flash(socket, :error, "Could not clear that job")}
133 + end
134 + end
135 +
136 + @doc """
137 + How long a job has been executing, and whether that has stopped being
138 + plausible.
139 +
140 + An import of a large feed genuinely takes many minutes and can go a long
141 + while between broadcasts, so elapsed time alone is a weak signal and the
142 + threshold is deliberately generous. Past it, the honest thing is to say the
143 + job looks stranded and let someone who knows decide, rather than either
144 + hiding the button or quietly killing real work.
145 + """
146 + @stranded_after_s 3_600
147 +
148 + def stranded?(%{state: "executing"} = job, progress) do
149 + not reporting?(job, progress) and elapsed(job.attempted_at) > @stranded_after_s
150 + end
151 +
152 + def stranded?(_job, _progress), do: false
153 +
154 + # A job that has said something recently is alive whatever the clock says.
155 + defp reporting?(job, progress) do
156 + case source_id(job) do
157 + nil -> false
158 + id -> Map.has_key?(progress, to_string(id))
159 + end
160 + end
161 +
162 + defp elapsed(nil), do: 0
163 + defp elapsed(%NaiveDateTime{} = at), do: elapsed(DateTime.from_naive!(at, "Etc/UTC"))
164 + defp elapsed(%DateTime{} = at), do: DateTime.diff(DateTime.utc_now(), at)
165 +
107 166 defp load(socket) do
108 167 sources =
109 168 Configuration.list_cfg_sources({:type, :gtfs})
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/work.html.heex
+25 −0
@@ -41,6 +41,18 @@
41 41 <p class="text-sm opacity-70">
42 42 <%= if running.progress, do: "#{running.progress.value}%", else: "picked up, no progress reported yet" %>
43 43 </p>
44 +
45 + <%!-- A job whose process died sits in `executing` for ever, and the
46 + card above would otherwise report it as work in progress
47 + indefinitely. --%>
48 + <% stuck = Enum.filter(@jobs, &stranded?(&1, @progress)) %>
49 + <div :if={stuck != []} class="alert alert-warning mt-2 text-sm">
50 + <i class="fa-solid fa-triangle-exclamation"></i>
51 + <span>
52 + <%= length(stuck) %> job<%= if length(stuck) > 1, do: "s" %> marked running with nothing
53 + reported for over an hour. Until cleared, the feed cannot be queued again.
54 + </span>
55 + </div>
44 56 <% else %>
45 57 <p class="text-sm opacity-70">Nothing is importing right now.</p>
46 58 <% end %>
@@ -106,6 +118,7 @@
106 118 <th>Started</th>
107 119 <th>Finished</th>
108 120 <th>Error</th>
121 + <th></th>
109 122 </tr>
110 123 </thead>
111 124 <tbody>
@@ -133,6 +146,18 @@
133 146 <%= ago(job.completed_at || job.discarded_at) %>
134 147 </td>
135 148 <td class="text-xs text-error"><%= last_error(job.errors) %></td>
149 + <td>
150 + <button
151 + :if={job.state == "executing"}
152 + class={"btn btn-xs #{if stranded?(job, @progress), do: "btn-warning", else: "btn-ghost"}"}
153 + phx-click="clear"
154 + phx-value-id={job.id}
155 + data-confirm="Clear this job? If the import is genuinely running it will be stopped."
156 + title="Give up on this job so the feed can be queued again"
157 + >
158 + clear
159 + </button>
160 + </td>
136 161 </tr>
137 162 <% end %>
138 163 </tbody>
modified apps/room_sanctum/test/room_sanctum_web/live/source_work_test.exs
+84 −0
@@ -47,6 +47,16 @@ defmodule RoomSanctumWeb.SourceWorkTest do
47 47 id
48 48 end
49 49
50 + defp jobs_for(id) do
51 + import Ecto.Query
52 +
53 + from(j in "oban_jobs",
54 + where: j.id == ^id,
55 + select: %{id: j.id, state: j.state, args: j.args, attempted_at: j.attempted_at}
56 + )
57 + |> Repo.all()
58 + end
59 +
50 60 test "the page lists this user's feeds", %{conn: conn, source: source} do
51 61 {:ok, _live, html} = live(conn, "/cfg/offerings/work")
52 62
@@ -178,6 +188,80 @@ defmodule RoomSanctumWeb.SourceWorkTest do
178 188 end
179 189 end
180 190
191 + describe "clearing a stuck job" do
192 + alias RoomSanctumWeb.SourceLive.Work
193 +
194 + defp executing_since(source_id, seconds_ago) do
195 + at =
196 + DateTime.utc_now()
197 + |> DateTime.add(-seconds_ago)
198 + |> DateTime.to_naive()
199 + |> NaiveDateTime.truncate(:second)
200 +
201 + job(source_id, "executing", %{attempt: 1, attempted_at: at})
202 + end
203 +
204 + test "a job running for a few minutes is not called stranded", %{source: source} do
205 + id = executing_since(source.id, 300)
206 + [job] = jobs_for(id)
207 +
208 + refute Work.stranded?(job, %{})
209 + end
210 +
211 + test "an hour of silence is", %{source: source} do
212 + id = executing_since(source.id, 7_200)
213 + [job] = jobs_for(id)
214 +
215 + assert Work.stranded?(job, %{})
216 + end
217 +
218 + test "a job still broadcasting is alive whatever the clock says", %{source: source} do
219 + id = executing_since(source.id, 7_200)
220 + [job] = jobs_for(id)
221 +
222 + progress = %{to_string(source.id) => %{step: :stop_times, value: 40, at: DateTime.utc_now()}}
223 +
224 + refute Work.stranded?(job, progress)
225 + end
226 +
227 + test "a job that is not executing is never stranded", %{source: source} do
228 + id = job(source.id, "completed")
229 + [job] = jobs_for(id)
230 +
231 + refute Work.stranded?(job, %{})
232 + end
233 +
234 + test "the page warns when a feed is blocked by one", %{conn: conn, source: source} do
235 + executing_since(source.id, 7_200)
236 +
237 + {:ok, _live, html} = live(conn, "/cfg/offerings/work")
238 +
239 + assert html =~ "marked running with nothing"
240 + # Why it matters, not just that it happened.
241 + assert html =~ "cannot be queued again"
242 + end
243 +
244 + test "clearing frees the feed", %{conn: conn, source: source} do
245 + id = executing_since(source.id, 7_200)
246 +
247 + {:ok, live, _html} = live(conn, "/cfg/offerings/work")
248 + render_click(live, "clear", %{"id" => to_string(id)})
249 +
250 + [job] = jobs_for(id)
251 + # Cancelled is outside the states ImportJob's uniqueness counts, which is
252 + # what lets the next import through.
253 + assert job.state == "cancelled"
254 + end
255 +
256 + test "only executing jobs offer the button", %{conn: conn, source: source} do
257 + job(source.id, "completed")
258 +
259 + {:ok, _live, html} = live(conn, "/cfg/offerings/work")
260 +
261 + refute html =~ ~s(phx-click="clear")
262 + end
263 + end
264 +
181 265 test "another user's feeds are not listed", %{conn: conn} do
182 266 {:ok, stranger} =
183 267 Accounts.register_user(%{

Parents: 448b406