Let a cleanup run clear the whole backlog

lib/bob/artifacts.ex

@@ -509,8 +509,8 @@ defmodule Bob.Artifacts do
     Enum.map(rows, fn [tag, archs] -> {tag, archs} end)
   end
 
-  # build_requests is small, so the reserved tag names are cheaper to collect
-  # and match against than to reconstruct from the component columns in SQL.
+  # build_requests is small, so collecting the reserved tag names and matching
+  # against the list is cheap.
   defp reserved_targets() do
     from(br in BuildRequest, where: br.state in ["pending", "completed"])
     |> Repo.all()
@@ -521,15 +521,19 @@ defmodule Bob.Artifacts do
 
   defp unreserved(query), do: where(query, [d], d.tag not in ^reserved_targets())
 
-  defp stale_per_arch(cutoff) do
-    from(d in DockerTag,
-      where: d.repo in @docker_cleanup_per_arch_repos and d.built_at < ^cutoff
-    )
+  def docker_cleanup_per_arch_repos(), do: @docker_cleanup_per_arch_repos
+
+  # Only the known per-arch repos, so a caller cannot point the cleanup at a
+  # manifest repo.
+  defp stale_per_arch(cutoff, repos) do
+    repos = Enum.filter(@docker_cleanup_per_arch_repos, &(&1 in repos))
+
+    from(d in DockerTag, where: d.repo in ^repos and d.built_at < ^cutoff)
   end
 
   @doc "Per-repo count of the tags a run would delete."
-  def count_stale_per_arch_tags(cutoff) do
-    stale_per_arch(cutoff)
+  def count_stale_per_arch_tags(cutoff, repos \\ @docker_cleanup_per_arch_repos) do
+    stale_per_arch(cutoff, repos)
     |> unreserved()
     |> group_by([d], d.repo)
     |> select([d], {d.repo, count(d.id)})
@@ -538,8 +542,8 @@ defmodule Bob.Artifacts do
   end
 
   @doc "Up to `limit` deletable `{repo, tag}` pairs."
-  def stale_per_arch_tags(cutoff, limit) do
-    stale_per_arch(cutoff)
+  def stale_per_arch_tags(cutoff, limit, repos \\ @docker_cleanup_per_arch_repos) do
+    stale_per_arch(cutoff, repos)
     |> unreserved()
     |> limit(^limit)
     |> select([d], {d.repo, d.tag})

lib/bob/docker_cleanup.ex

@@ -3,9 +3,13 @@ defmodule Bob.DockerCleanup do
   Deletes per-arch Docker Hub tags older than 30 days. Tags reserved by a build
   request are kept. The manifest repos are not pruned.
 
-  `run/1` deletes one batch and is what the nightly job calls; `drain/1` loops
-  until the backlog is empty. `:docker_cleanup_mode` gates whether `run/1`
-  deletes or only reports.
+  A live run keeps taking batches until one deletes nothing, so it clears the
+  whole backlog rather than a single batch. What bounds it is the job's timeout.
+  `:docker_cleanup_mode` gates whether it deletes or only reports.
+
+  `:repos` narrows a run to some of the per-arch repos. Docker Hub rate limits
+  deletes per source IP, so pointing one node at each repo doubles throughput,
+  where two nodes on the same repo just race for the same tags.
   """
 
   require Logger
@@ -16,7 +20,7 @@ defmodule Bob.DockerCleanup do
   # tags the checker still expects. Asserted in docker_cleanup_test.
   @per_arch_max_age_days 30
 
-  # Sized to finish inside Bob.Runner's three-hour job timeout.
+  # Candidate query page size.
   @default_batch 10_000
 
   # Rows are committed per chunk so a killed run keeps its progress.
@@ -29,42 +33,21 @@ defmodule Bob.DockerCleanup do
   @default_concurrency 25
 
   def run(opts \\ []) do
+    repos = Keyword.get(opts, :repos, Artifacts.docker_cleanup_per_arch_repos())
+
     case Keyword.get(opts, :mode, configured_mode()) do
-      :dry_run -> dry_run()
-      :live -> live(opts)
+      :dry_run -> dry_run(repos)
+      :live -> live(opts, repos)
     end
   end
 
-  @doc """
-  Runs batches until one deletes nothing. Always deletes, whatever
-  `:docker_cleanup_mode` says, and is not the scheduled path — the job runner
-  kills anything past three hours, so start this from a supervised task.
-  """
-  def drain(opts \\ []) do
-    opts = Keyword.put_new(opts, :mode, :live)
-    batch = Keyword.get(opts, :limit, @default_batch)
-
-    Stream.repeatedly(fn -> run(Keyword.put(opts, :limit, batch)) end)
-    |> Enum.reduce_while(0, fn {:live, deleted}, total ->
-      total = total + deleted
-
-      if deleted == 0 do
-        Logger.info("DOCKER CLEANUP drain finished, deleted #{total} tag(s)")
-        {:halt, total}
-      else
-        Logger.info("DOCKER CLEANUP drain progress, deleted #{total} tag(s) so far")
-        {:cont, total}
-      end
-    end)
-  end
-
   defp configured_mode(), do: Application.get_env(:bob, :docker_cleanup_mode, :dry_run)
 
-  defp dry_run() do
-    per_arch = Artifacts.count_stale_per_arch_tags(per_arch_cutoff())
+  defp dry_run(repos) do
+    per_arch = Artifacts.count_stale_per_arch_tags(per_arch_cutoff(), repos)
     backlog = per_arch |> Map.values() |> Enum.sum()
 
-    # Report both: the backlog is every candidate, a run stops at the batch.
+    # The backlog is every candidate; a scheduled run stops at the batch.
     Logger.info(
       "DOCKER CLEANUP dry-run: per-arch built_at < #{@per_arch_max_age_days}d -> " <>
         "#{format_counts(per_arch)}; one scheduled run would delete " <>
@@ -74,15 +57,22 @@ defmodule Bob.DockerCleanup do
     {:dry_run, %{per_arch: per_arch}}
   end
 
-  defp live(opts) do
+  defp live(opts, repos) do
     deleter = Keyword.get(opts, :deleter, &Bob.DockerHub.delete_tag/2)
     limit = Keyword.get(opts, :limit, @default_batch)
     cutoff = per_arch_cutoff()
 
-    candidates = Artifacts.stale_per_arch_tags(cutoff, limit)
-
-    deleted = delete(candidates, deleter, cutoff)
-    Logger.info("DOCKER CLEANUP deleted #{deleted}/#{length(candidates)} tag(s)")
+    deleted =
+      Stream.repeatedly(fn ->
+        cutoff
+        |> Artifacts.stale_per_arch_tags(limit, repos)
+        |> delete(deleter, cutoff)
+      end)
+      |> Enum.reduce_while(0, fn batch, total ->
+        if batch == 0, do: {:halt, total}, else: {:cont, total + batch}
+      end)
+
+    Logger.info("DOCKER CLEANUP deleted #{deleted} tag(s) from #{Enum.join(repos, ", ")}")
     {:live, deleted}
   end
 
@@ -97,8 +87,7 @@ defmodule Bob.DockerCleanup do
       Artifacts.delete_docker_tags(confirmed)
       deleted = deleted + length(confirmed)
 
-      # Whole failed chunks, not individual errors: concurrent deletes have no
-      # "consecutive", and a flaky tag shouldn't trip the abort.
+      # Counted per chunk so a flaky tag does not trip the abort.
       dead_chunks = if failed > 0 and confirmed == [], do: dead_chunks + 1, else: 0
 
       if dead_chunks >= @dead_chunk_ceiling do

lib/bob/job.ex

@@ -1,5 +1,28 @@
 defmodule Bob.Job do
   @type args :: [term()]
 
+  # A job still running after this is wedged, so the runner reaps it and the
+  # maintenance sweep requeues it. Jobs that legitimately run longer override
+  # timeout/0.
+  @default_timeout 3 * 60 * 60 * 1000
+
   @callback run(args()) :: term()
+  @callback timeout() :: pos_integer()
+
+  @optional_callbacks timeout: 0
+
+  def default_timeout(), do: @default_timeout
+
+  @doc "How long `key` may run before it is treated as wedged, in milliseconds."
+  def timeout({module, _key}), do: timeout(module)
+
+  # ensure_loaded? first: function_exported?/3 answers false for a module that
+  # has not been loaded yet, which silently hands every job the default.
+  def timeout(module) do
+    if Code.ensure_loaded?(module) and function_exported?(module, :timeout, 0) do
+      module.timeout()
+    else
+      @default_timeout
+    end
+  end
 end

lib/bob/job/docker_cleanup.ex

@@ -3,6 +3,10 @@ defmodule Bob.Job.DockerCleanup do
     Bob.DockerCleanup.run()
   end
 
+  # A live run clears the whole backlog, which takes days at Docker Hub's delete
+  # rate. The next night's run is dedup'd away while this one is still going.
+  def timeout(), do: 23 * 60 * 60 * 1000
+
   def priority(), do: 1
   def weight(), do: 1
   def concurrency(), do: :shared

lib/bob/queue/maintenance.ex

@@ -15,7 +15,6 @@ defmodule Bob.Queue.Maintenance do
   alias Bob.Queue.{Job, Failure}
 
   @interval_seconds 60
-  @job_timeout_seconds 3 * 60 * 60
   # A job stuck in running this long means its node died hard (OOM, node loss),
   # so the work itself is not suspect — requeue it. The cap stops a job that
   # reliably kills its node or genuinely exceeds the timeout from looping.
@@ -67,12 +66,23 @@ defmodule Bob.Queue.Maintenance do
 
   defp sweep_stale_running() do
     now = DateTime.utc_now()
-    cutoff = DateTime.add(now, -@job_timeout_seconds, :second)
 
-    stale =
+    # Pre-filter on the shortest timeout any job can have, then keep only those
+    # past their own.
+    cutoff = DateTime.add(now, -div(Bob.Job.default_timeout(), 1000), :second)
+
+    stale_ids =
       from(j in Job,
-        where: j.state == "running" and not is_nil(j.started_at) and j.started_at < ^cutoff
+        where: j.state == "running" and not is_nil(j.started_at) and j.started_at < ^cutoff,
+        select: {j.id, j.module_key, j.started_at}
       )
+      |> Repo.all()
+      |> Enum.filter(fn {_id, module_key, started_at} ->
+        DateTime.diff(now, started_at, :millisecond) > Bob.Job.timeout(module_key)
+      end)
+      |> Enum.map(fn {id, _module_key, _started_at} -> id end)
+
+    stale = from(j in Job, where: j.id in ^stale_ids)
 
     {requeued, _} =
       Repo.update_all(

lib/bob/runner.ex

@@ -9,10 +9,8 @@ defmodule Bob.Runner do
 
   # A build that hangs (e.g. a wedged `docker build`) blocks its task forever and
   # keeps counting its weight against the shared budget, so the agent stops
-  # pulling new builds. Reap a task that outlives the master's stale-job timeout
-  # (mirrors Bob.Queue.Maintenance @job_timeout_seconds) so the slot is freed and
-  # the job is reported failed.
-  @job_timeout 3 * 60 * 60 * 1000
+  # pulling new builds. Reap a task that outlives its job's timeout so the slot
+  # is freed and the job is reported failed.
 
   def start_link([]) do
     GenServer.start_link(__MODULE__, new_state(), name: __MODULE__)
@@ -92,7 +90,7 @@ defmodule Bob.Runner do
     end
   end
 
-  # A task still running past @job_timeout is wedged (e.g. a hung `docker build`).
+  # A task still running past its timeout is wedged (e.g. a hung `docker build`).
   # Kill it so it stops leaking its weight, fail the job, and pull fresh work.
   # The kill triggers an abnormal :DOWN, but the row is already gone so that
   # clause no-ops.
@@ -202,7 +200,7 @@ defmodule Bob.Runner do
   defp start_job(id, key, args, state) do
     Logger.info("STARTING #{inspect(key)} #{inspect(args)}")
     task = Task.Supervisor.async(Bob.Tasks, fn -> run_task(key, args) end)
-    timer = Process.send_after(self(), {:job_timeout, task.ref}, @job_timeout)
+    timer = Process.send_after(self(), {:job_timeout, task.ref}, Bob.Job.timeout(key))
     put_in(state.tasks[task.ref], {key, args, id, task.pid, timer})
   end
 

test/bob/docker_cleanup_test.exs

@@ -131,20 +131,20 @@ defmodule Bob.DockerCleanupTest do
                [{"1.26.0-erlang-27.0-ubuntu-noble-20250101", ["amd64"]}]
     end
 
-    test "respects the batch limit so a run deletes a bounded slice" do
+    test "the batch limit pages the candidate query without capping the run" do
       deleter = fn _repo, _tag -> :ok end
 
       for n <- 1..3 do
         Artifacts.add_docker_tag(
           "hexpm/elixir-amd64",
-          "2#{n}.0-ubuntu-noble-20250101",
+          "1.2#{n}.0-erlang-27.0-ubuntu-noble-20250101",
           ["amd64"],
           old()
         )
       end
 
-      assert {:live, 2} = DockerCleanup.run(mode: :live, deleter: deleter, limit: 2)
-      assert length(Artifacts.docker_tags("hexpm/elixir-amd64")) == 1
+      assert {:live, 3} = DockerCleanup.run(mode: :live, deleter: deleter, limit: 2)
+      assert Artifacts.docker_tags("hexpm/elixir-amd64") == []
     end
 
     test "an ancient manifest tag is never a candidate, whatever the limit" do
@@ -216,8 +216,63 @@ defmodule Bob.DockerCleanupTest do
     end
   end
 
-  describe "drain/1" do
-    test "keeps going past the batch limit until nothing is left" do
+  describe "run/1 scoped to some repos" do
+    setup do
+      for repo <- ~w(hexpm/elixir-amd64 hexpm/elixir-arm64) do
+        Artifacts.add_docker_tag(
+          repo,
+          "1.18.0-erlang-27.0-ubuntu-noble-20250101",
+          ["amd64"],
+          old()
+        )
+      end
+
+      :ok
+    end
+
+    test "deletes only from the named repo" do
+      test = self()
+      deleter = fn repo, _tag -> send(test, {:deleted, repo}) && :ok end
+
+      assert {:live, 1} =
+               DockerCleanup.run(
+                 mode: :live,
+                 deleter: deleter,
+                 repos: ["hexpm/elixir-amd64"]
+               )
+
+      assert_received {:deleted, "hexpm/elixir-amd64"}
+      refute_received {:deleted, "hexpm/elixir-arm64"}
+      assert Artifacts.docker_tags("hexpm/elixir-arm64") != []
+    end
+
+    test "the dry run counts only the named repo" do
+      assert {:dry_run, %{per_arch: counts}} =
+               DockerCleanup.run(mode: :dry_run, repos: ["hexpm/elixir-arm64"])
+
+      assert counts == %{"hexpm/elixir-arm64" => 1}
+    end
+
+    # A repo outside the list would be the images users pull.
+    test "a repo outside the per-arch list is ignored, not deleted from" do
+      deleter = fn _repo, _tag -> :ok end
+
+      Artifacts.add_docker_tag(
+        "hexpm/elixir",
+        "1.17.0-erlang-27.0-ubuntu-noble-20250101",
+        ["amd64", "arm64"],
+        ancient()
+      )
+
+      assert {:live, 0} =
+               DockerCleanup.run(mode: :live, deleter: deleter, repos: ["hexpm/elixir"])
+
+      assert Artifacts.docker_tags("hexpm/elixir") != []
+    end
+  end
+
+  describe "run/1 clears the whole backlog" do
+    test "keeps taking batches until nothing is left" do
       deleter = fn _repo, _tag -> :ok end
 
       for n <- 1..7 do
@@ -229,8 +284,7 @@ defmodule Bob.DockerCleanupTest do
         )
       end
 
-      # A single run of this batch size would stop at 2; the drain loops.
-      assert DockerCleanup.drain(limit: 2, deleter: deleter) == 7
+      assert {:live, 7} = DockerCleanup.run(mode: :live, limit: 2, deleter: deleter)
       assert Artifacts.docker_tags("hexpm/elixir-amd64") == []
     end
 
@@ -244,7 +298,7 @@ defmodule Bob.DockerCleanupTest do
         old()
       )
 
-      assert DockerCleanup.drain(deleter: deleter) == 0
+      assert {:live, 0} = DockerCleanup.run(mode: :live, deleter: deleter)
       assert length(Artifacts.docker_tags("hexpm/elixir-amd64")) == 1
     end
 
@@ -275,7 +329,7 @@ defmodule Bob.DockerCleanupTest do
         builds_count: 0
       })
 
-      assert DockerCleanup.drain(limit: 2, deleter: deleter) == 0
+      assert {:live, 0} = DockerCleanup.run(mode: :live, limit: 2, deleter: deleter)
       assert length(Artifacts.docker_tags("hexpm/elixir-amd64")) == 2
     end
   end