Improve performance of job counting in maintenance
- Use a single query for all workers - Replace the manual Ecto query against Oban.Job with Oban.Job.query/1
This commit is contained in:
@@ -10,20 +10,23 @@ defmodule MusicLibrary.Maintenance do
|
|||||||
alias MusicLibrary.BackgroundRepo
|
alias MusicLibrary.BackgroundRepo
|
||||||
alias MusicLibrary.Repo
|
alias MusicLibrary.Repo
|
||||||
|
|
||||||
|
@active_states ~w(available scheduled executing retryable)
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Counts active Oban jobs for the given worker module name.
|
Returns a map of worker module names to their count of active Oban jobs.
|
||||||
|
|
||||||
Active jobs are those in "available", "scheduled", "executing", or "retryable" states.
|
Active jobs are those in "available", "scheduled", "executing", or "retryable" states.
|
||||||
"""
|
"""
|
||||||
@spec count_active_jobs(String.t()) :: non_neg_integer() | nil
|
@spec count_active_jobs_by_worker() :: %{String.t() => non_neg_integer()}
|
||||||
def count_active_jobs(worker) do
|
def count_active_jobs_by_worker do
|
||||||
query =
|
q =
|
||||||
from j in Oban.Job,
|
from j in subquery(Oban.Job.query(state: @active_states)),
|
||||||
where: j.worker == ^worker,
|
group_by: j.worker,
|
||||||
where: j.state in ["available", "scheduled", "executing", "retryable"],
|
select: {j.worker, count(j.id)}
|
||||||
select: count(j.id)
|
|
||||||
|
|
||||||
BackgroundRepo.one(query)
|
q
|
||||||
|
|> BackgroundRepo.all()
|
||||||
|
|> Map.new()
|
||||||
end
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
|
|||||||
@@ -305,30 +305,32 @@ defmodule MusicLibraryWeb.MaintenanceLive.Index do
|
|||||||
end
|
end
|
||||||
|
|
||||||
defp assign_job_counts(socket) do
|
defp assign_job_counts(socket) do
|
||||||
|
counts = Maintenance.count_active_jobs_by_worker()
|
||||||
|
|
||||||
socket
|
socket
|
||||||
|> assign(
|
|> assign(
|
||||||
:refresh_records_musicbrainz_jobs,
|
:refresh_records_musicbrainz_jobs,
|
||||||
Maintenance.count_active_jobs("MusicLibrary.Worker.RecordRefreshMusicBrainzData")
|
Map.get(counts, "MusicLibrary.Worker.RecordRefreshMusicBrainzData", 0)
|
||||||
)
|
)
|
||||||
|> assign(
|
|> assign(
|
||||||
:generate_record_embeddings_jobs,
|
:generate_record_embeddings_jobs,
|
||||||
Maintenance.count_active_jobs("MusicLibrary.Worker.GenerateRecordEmbedding")
|
Map.get(counts, "MusicLibrary.Worker.GenerateRecordEmbedding", 0)
|
||||||
)
|
)
|
||||||
|> assign(
|
|> assign(
|
||||||
:refresh_artists_musicbrainz_jobs,
|
:refresh_artists_musicbrainz_jobs,
|
||||||
Maintenance.count_active_jobs("MusicLibrary.Worker.ArtistRefreshMusicBrainzData")
|
Map.get(counts, "MusicLibrary.Worker.ArtistRefreshMusicBrainzData", 0)
|
||||||
)
|
)
|
||||||
|> assign(
|
|> assign(
|
||||||
:refresh_artists_discogs_jobs,
|
:refresh_artists_discogs_jobs,
|
||||||
Maintenance.count_active_jobs("MusicLibrary.Worker.ArtistRefreshDiscogsData")
|
Map.get(counts, "MusicLibrary.Worker.ArtistRefreshDiscogsData", 0)
|
||||||
)
|
)
|
||||||
|> assign(
|
|> assign(
|
||||||
:refresh_artists_wikipedia_jobs,
|
:refresh_artists_wikipedia_jobs,
|
||||||
Maintenance.count_active_jobs("MusicLibrary.Worker.ArtistRefreshWikipediaData")
|
Map.get(counts, "MusicLibrary.Worker.ArtistRefreshWikipediaData", 0)
|
||||||
)
|
)
|
||||||
|> assign(
|
|> assign(
|
||||||
:refresh_artists_lastfm_jobs,
|
:refresh_artists_lastfm_jobs,
|
||||||
Maintenance.count_active_jobs("MusicLibrary.Worker.FetchArtistLastFmData")
|
Map.get(counts, "MusicLibrary.Worker.FetchArtistLastFmData", 0)
|
||||||
)
|
)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ defmodule MusicLibrary.MaintenanceTest do
|
|||||||
|
|
||||||
import MusicLibrary.ScrobbledTracksFixtures
|
import MusicLibrary.ScrobbledTracksFixtures
|
||||||
|
|
||||||
|
alias MusicLibrary.BackgroundRepo
|
||||||
alias MusicLibrary.Maintenance
|
alias MusicLibrary.Maintenance
|
||||||
|
|
||||||
describe "optimize/0" do
|
describe "optimize/0" do
|
||||||
@@ -11,9 +12,47 @@ defmodule MusicLibrary.MaintenanceTest do
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
describe "count_active_jobs/1" do
|
describe "count_active_jobs_by_worker/0" do
|
||||||
test "returns 0 for a worker with no jobs" do
|
@worker_a "Test.Worker.Alpha"
|
||||||
assert Maintenance.count_active_jobs("MusicLibrary.Worker.NonExistent") == 0
|
@worker_b "Test.Worker.Beta"
|
||||||
|
|
||||||
|
test "returns an empty map when there are no jobs" do
|
||||||
|
assert Maintenance.count_active_jobs_by_worker() == %{}
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns counts grouped by worker for active jobs" do
|
||||||
|
%{id: "1"}
|
||||||
|
|> Oban.Job.new(worker: @worker_a)
|
||||||
|
|> Oban.insert!()
|
||||||
|
|
||||||
|
%{id: "2"}
|
||||||
|
|> Oban.Job.new(worker: @worker_a)
|
||||||
|
|> Oban.insert!()
|
||||||
|
|
||||||
|
%{id: "3"}
|
||||||
|
|> Oban.Job.new(worker: @worker_b)
|
||||||
|
|> Oban.insert!()
|
||||||
|
|
||||||
|
counts = Maintenance.count_active_jobs_by_worker()
|
||||||
|
|
||||||
|
assert counts[@worker_a] == 2
|
||||||
|
assert counts[@worker_b] == 1
|
||||||
|
end
|
||||||
|
|
||||||
|
test "only counts jobs in active states" do
|
||||||
|
%{id: "1"}
|
||||||
|
|> Oban.Job.new(worker: @worker_a)
|
||||||
|
|> Oban.insert!()
|
||||||
|
|> then(fn job ->
|
||||||
|
BackgroundRepo.update_all(
|
||||||
|
from(j in Oban.Job, where: j.id == ^job.id),
|
||||||
|
set: [state: "completed"]
|
||||||
|
)
|
||||||
|
end)
|
||||||
|
|
||||||
|
counts = Maintenance.count_active_jobs_by_worker()
|
||||||
|
|
||||||
|
assert Map.get(counts, @worker_a, 0) == 0
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user