diff --git a/Gemfile b/Gemfile index e9a66f75..8a3a4033 100644 --- a/Gemfile +++ b/Gemfile @@ -5,3 +5,8 @@ git_source(:github) { |repo| "https://github.com/#{repo}.git" } gemspec gem "capybara", github: "teamcapybara/capybara" +# Use `path: "../solid_queue"` when co-developing locally +gem "solid_queue", github: "jpcamara/solid_queue", branch: "batch-review-fixes" + +# minitest 6 is incompatible with railties 7.2 line filtering +gem "minitest", "~> 5.25" diff --git a/Gemfile.lock b/Gemfile.lock index 7246e279..ae7117e4 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,3 +1,16 @@ +GIT + remote: https://github.com/jpcamara/solid_queue.git + revision: cbfa930444fc85dad2065df10ebd4dea85fdb933 + branch: batch-review-fixes + specs: + solid_queue (1.5.0) + activejob (>= 7.1) + activerecord (>= 7.1) + concurrent-ruby (>= 1.3.1) + fugit (~> 1.11) + railties (>= 7.1) + thor (>= 1.3.1) + GIT remote: https://github.com/teamcapybara/capybara.git revision: c0cbf4024c1abd48b0c22c2930e7b05af58ab284 @@ -76,8 +89,8 @@ GEM addressable (2.8.6) public_suffix (>= 2.0.2, < 6.0) ast (2.4.2) - base64 (0.2.0) - benchmark (0.3.0) + base64 (0.3.0) + benchmark (0.5.0) better_html (2.1.1) actionview (>= 6.0) activesupport (>= 6.0) @@ -85,41 +98,45 @@ GEM erubi (~> 1.4) parser (>= 2.4) smart_properties - bigdecimal (3.1.8) + bigdecimal (4.1.2) builder (3.3.0) - concurrent-ruby (1.3.4) - connection_pool (2.4.1) - crass (1.0.6) + concurrent-ruby (1.3.8) + connection_pool (3.0.2) + crass (1.0.7) debug (1.9.1) irb (~> 1.10) reline (>= 0.3.8) - drb (2.2.1) - erubi (1.13.0) - et-orbi (1.2.11) + drb (2.2.3) + erb (6.0.6) + erubi (1.13.1) + et-orbi (1.4.0) tzinfo - fugit (1.11.1) - et-orbi (~> 1, >= 1.2.11) + fugit (1.13.0) + et-orbi (~> 1.4) raabro (~> 1.4) - globalid (1.2.1) + globalid (1.4.0) activesupport (>= 6.1) - i18n (1.14.6) + i18n (1.15.2) concurrent-ruby (~> 1.0) importmap-rails (2.0.1) actionpack (>= 6.0.0) activesupport (>= 6.0.0) railties (>= 6.0.0) - io-console (0.7.2) - irb (1.14.1) + io-console (0.8.2) + irb (1.18.0) + pp (>= 0.6.0) + prism (>= 1.3.0) rdoc (>= 4.0.0) reline (>= 0.4.2) json (2.7.1) - logger (1.6.1) - loofah (2.23.1) + logger (1.7.0) + loofah (2.25.2) crass (~> 1.0.2) nokogiri (>= 1.12.0) matrix (0.4.2) mini_mime (1.1.5) - minitest (5.25.1) + mini_portile2 (2.8.9) + minitest (5.27.0) mocha (2.1.0) ruby2_keywords (>= 0.0.5) mono_logger (1.1.2) @@ -127,52 +144,56 @@ GEM mustermann (3.0.3) ruby2_keywords (~> 0.0.1) nio4r (2.7.4) - nokogiri (1.16.7-aarch64-linux) + nokogiri (1.19.4) + mini_portile2 (~> 2.8.2) racc (~> 1.4) - nokogiri (1.16.7-arm-linux) + nokogiri (1.19.4-aarch64-linux-gnu) racc (~> 1.4) - nokogiri (1.16.7-arm64-darwin) + nokogiri (1.19.4-arm-linux-gnu) racc (~> 1.4) - nokogiri (1.16.7-x86-linux) + nokogiri (1.19.4-arm64-darwin) racc (~> 1.4) - nokogiri (1.16.7-x86_64-darwin) + nokogiri (1.19.4-x86_64-darwin) racc (~> 1.4) - nokogiri (1.16.7-x86_64-linux) + nokogiri (1.19.4-x86_64-linux-gnu) racc (~> 1.4) parallel (1.24.0) parser (3.3.0.5) ast (~> 2.4.1) racc + pp (0.6.4) + prettyprint + prettyprint (0.2.0) + prism (1.9.0) propshaft (1.1.0) actionpack (>= 7.0.0) activesupport (>= 7.0.0) rack railties (>= 7.0.0) - psych (5.1.2) - stringio public_suffix (5.0.4) puma (6.4.3) nio4r (~> 2.0) - raabro (1.4.0) + raabro (1.5.0) racc (1.8.1) - rack (3.1.8) + rack (3.1.21) rack-protection (4.1.0) base64 (>= 0.1.0) logger (>= 1.6.0) rack (>= 3.0.0, < 4) - rack-session (2.0.0) + rack-session (2.1.2) + base64 (>= 0.1.0) rack (>= 3.0.0) - rack-test (2.1.0) + rack-test (2.2.0) rack (>= 1.3) - rackup (2.2.0) + rackup (2.3.1) rack (>= 3) - rails-dom-testing (2.2.0) + rails-dom-testing (2.3.0) activesupport (>= 5.0.0) minitest nokogiri (>= 1.6) - rails-html-sanitizer (1.6.0) - loofah (~> 2.21) - nokogiri (~> 1.14) + rails-html-sanitizer (1.7.1) + loofah (~> 2.25, >= 2.25.2) + nokogiri (>= 1.15.7, != 1.16.7, != 1.16.6, != 1.16.5, != 1.16.4, != 1.16.3, != 1.16.2, != 1.16.1, != 1.16.0.rc1, != 1.16.0) railties (7.2.2) actionpack (= 7.2.2) activesupport (= 7.2.2) @@ -182,9 +203,16 @@ GEM thor (~> 1.0, >= 1.2.2) zeitwerk (~> 2.6) rainbow (3.1.1) - rake (13.2.1) - rdoc (6.7.0) - psych (>= 4.0.0) + rake (13.4.2) + rbs (4.1.0) + logger + prism (>= 1.6.0) + tsort + rdoc (8.0.0) + erb + prism (>= 1.6.0) + rbs (>= 4.0.0) + tsort redis (5.1.0) redis-client (>= 0.17.0) redis-client (0.21.0) @@ -192,7 +220,7 @@ GEM redis-namespace (1.11.0) redis (>= 4) regexp_parser (2.9.0) - reline (0.5.10) + reline (0.6.3) io-console (~> 0.5) resque (2.6.0) mono_logger (~> 1.0) @@ -234,7 +262,7 @@ GEM ruby-progressbar (1.13.0) ruby2_keywords (0.0.5) rubyzip (2.3.2) - securerandom (0.3.1) + securerandom (0.4.1) selenium-webdriver (4.18.1) base64 (~> 0.2) rexml (~> 3.2, >= 3.2.5) @@ -248,25 +276,18 @@ GEM rack-session (>= 2.0.0, < 3) tilt (~> 2.0) smart_properties (1.17.0) - solid_queue (1.0.1) - activejob (>= 7.1) - activerecord (>= 7.1) - concurrent-ruby (>= 1.3.1) - fugit (~> 1.11.0) - railties (>= 7.1) - thor (~> 1.3.1) - sqlite3 (2.2.0-aarch64-linux-gnu) - sqlite3 (2.2.0-arm-linux-gnu) - sqlite3 (2.2.0-arm64-darwin) - sqlite3 (2.2.0-x86-linux-gnu) - sqlite3 (2.2.0-x86_64-darwin) - sqlite3 (2.2.0-x86_64-linux-gnu) + sqlite3 (2.9.5-aarch64-linux-gnu) + sqlite3 (2.9.5-arm-linux-gnu) + sqlite3 (2.9.5-arm64-darwin) + sqlite3 (2.9.5-x86-linux-gnu) + sqlite3 (2.9.5-x86_64-darwin) + sqlite3 (2.9.5-x86_64-linux-gnu) stimulus-rails (1.3.3) railties (>= 6.0.0) - stringio (3.1.1) - thor (1.3.2) + thor (1.5.0) tilt (2.4.0) - timeout (0.4.1) + timeout (0.6.1) + tsort (0.2.0) turbo-rails (2.0.5) actionpack (>= 6.0.0) activejob (>= 6.0.0) @@ -274,14 +295,14 @@ GEM tzinfo (2.0.6) concurrent-ruby (~> 1.0) unicode-display_width (2.5.0) - useragent (0.16.10) + useragent (0.16.11) websocket (1.2.10) websocket-driver (0.7.6) websocket-extensions (>= 0.1.0) websocket-extensions (0.1.5) xpath (3.2.0) nokogiri (~> 1.8) - zeitwerk (2.7.1) + zeitwerk (2.8.2) PLATFORMS aarch64-linux @@ -295,6 +316,7 @@ DEPENDENCIES better_html capybara! debug + minitest (~> 5.25) mission_control-jobs! mocha propshaft @@ -307,7 +329,7 @@ DEPENDENCIES rubocop-performance rubocop-rails-omakase selenium-webdriver - solid_queue (~> 1.0.1) + solid_queue! sqlite3 BUNDLED WITH diff --git a/app/controllers/concerns/mission_control/jobs/adapter_features.rb b/app/controllers/concerns/mission_control/jobs/adapter_features.rb index e4e09c72..a55b136e 100644 --- a/app/controllers/concerns/mission_control/jobs/adapter_features.rb +++ b/app/controllers/concerns/mission_control/jobs/adapter_features.rb @@ -2,7 +2,7 @@ module MissionControl::Jobs::AdapterFeatures extend ActiveSupport::Concern included do - helper_method :supported_job_statuses, :queue_pausing_supported?, :workers_exposed?, :recurring_tasks_supported? + helper_method :supported_job_statuses, :queue_pausing_supported?, :workers_exposed?, :recurring_tasks_supported?, :batches_supported? end private @@ -21,4 +21,8 @@ def workers_exposed? def recurring_tasks_supported? MissionControl::Jobs::Current.server.queue_adapter.supports_recurring_tasks? end + + def batches_supported? + MissionControl::Jobs::Current.server.queue_adapter.supports_batches? + end end diff --git a/app/controllers/concerns/mission_control/jobs/not_found_redirections.rb b/app/controllers/concerns/mission_control/jobs/not_found_redirections.rb index 672230f1..7230d462 100644 --- a/app/controllers/concerns/mission_control/jobs/not_found_redirections.rb +++ b/app/controllers/concerns/mission_control/jobs/not_found_redirections.rb @@ -26,6 +26,8 @@ def best_location_for_job_relation(job_relation) def best_location_for_resource_not_found_error(error) if error.message.match?(/recurring task/i) application_recurring_tasks_path(@application) + elsif error.message.match?(/batch/i) + application_batches_path(@application) else root_url end diff --git a/app/controllers/mission_control/jobs/batches_controller.rb b/app/controllers/mission_control/jobs/batches_controller.rb new file mode 100644 index 00000000..21f0dad3 --- /dev/null +++ b/app/controllers/mission_control/jobs/batches_controller.rb @@ -0,0 +1,59 @@ +class MissionControl::Jobs::BatchesController < MissionControl::Jobs::ApplicationController + UNFINISHED_JOB_STATUSES = %i[ pending in_progress blocked scheduled ] + BATCHES_STATUSES = %w[ finished unfinished failed ] + + before_action :ensure_supported_batches + before_action :set_batch, only: :show + + def index + @batches_page = MissionControl::Jobs::Page.new(batches, page: params[:page].to_i) + end + + def show + @jobs_page = MissionControl::Jobs::Page.new(@batch.jobs.with_status(jobs_status.to_sym), page: params[:page].to_i) + end + + private + def ensure_supported_batches + unless batches_supported? + redirect_to root_url, alert: "This server doesn't support batches" + end + end + + def set_batch + @batch = MissionControl::Jobs::Current.server.find_batch(params[:id]) + end + + def batches + MissionControl::Jobs::Current.server.batches(status: batches_status&.to_sym) + end + + helper_method :batches_status, :batches_filter_param, :jobs_status + + # Default to unfinished: finished/all populations can be multi-million-row, + # and only unfinished batches are actionable. + def batches_status + requested = params.fetch(:batches_status, "unfinished") + requested.presence_in(BATCHES_STATUSES) unless requested == "all" + end + + def batches_filter_param + { batches_status: batches_status }.compact + end + + def jobs_status + if status = params[:jobs_status].presence&.to_sym.presence_in(supported_job_statuses) + status.to_s + elsif @batch.failed? + "failed" + elsif @batch.finished? + "finished" + else + default_unfinished_jobs_status + end + end + + def default_unfinished_jobs_status + UNFINISHED_JOB_STATUSES.find { |status| @batch.public_send("#{status}_jobs") > 0 }&.to_s || "pending" + end +end diff --git a/app/helpers/mission_control/jobs/interface_helper.rb b/app/helpers/mission_control/jobs/interface_helper.rb index 68e611e0..7daf39ac 100644 --- a/app/helpers/mission_control/jobs/interface_helper.rb +++ b/app/helpers/mission_control/jobs/interface_helper.rb @@ -20,4 +20,13 @@ def modifier_for_status(status) else "is-primary is-light" end end + + def modifier_for_batch_status(status) + case status.to_s + when "completed" then "is-success" + when "failed" then "is-danger" + when "enqueued" then "is-info" + else "is-light" + end + end end diff --git a/app/helpers/mission_control/jobs/navigation_helper.rb b/app/helpers/mission_control/jobs/navigation_helper.rb index d3d81a80..e4077dfd 100644 --- a/app/helpers/mission_control/jobs/navigation_helper.rb +++ b/app/helpers/mission_control/jobs/navigation_helper.rb @@ -9,6 +9,7 @@ def navigation_sections sections[:workers] = [ "Workers", application_workers_path(@application) ] if workers_exposed? sections[:recurring_tasks] = [ "Recurring tasks", application_recurring_tasks_path(@application) ] if recurring_tasks_supported? + sections[:batches] = [ "Batches", application_batches_path(@application) ] if batches_supported? end end diff --git a/app/models/mission_control/jobs/batch.rb b/app/models/mission_control/jobs/batch.rb new file mode 100644 index 00000000..409440f7 --- /dev/null +++ b/app/models/mission_control/jobs/batch.rb @@ -0,0 +1,38 @@ +class MissionControl::Jobs::Batch + include ActiveModel::Model + + attr_accessor :id, :description, :status, :total_jobs, :completed_jobs, :failed_jobs, + :pending_jobs, :in_progress_jobs, :blocked_jobs, :scheduled_jobs, :progress_percentage, + :metadata, :enqueued_at, :finished_at + + def initialize(queue_adapter: ActiveJob::Base.queue_adapter, **kwargs) + @queue_adapter = queue_adapter + super(**kwargs) + end + + def jobs + ActiveJob::JobsRelation.new(queue_adapter: queue_adapter).where(batch_id: id) + end + + def job_counts + { + finished: completed_jobs, + failed: failed_jobs, + pending: pending_jobs, + in_progress: in_progress_jobs, + blocked: blocked_jobs, + scheduled: scheduled_jobs + } + end + + def finished? + finished_at.present? + end + + def failed? + status == :failed + end + + private + attr_reader :queue_adapter +end diff --git a/app/models/mission_control/jobs/batches_relation.rb b/app/models/mission_control/jobs/batches_relation.rb new file mode 100644 index 00000000..a54ff545 --- /dev/null +++ b/app/models/mission_control/jobs/batches_relation.rb @@ -0,0 +1,91 @@ +# A relation of batches. +# +# Relations are enumerable, so you can use +Enumerable+ methods on them. +# Notice however that using these methods will imply loading all the relation +# in memory, which could introduce performance concerns. +class MissionControl::Jobs::BatchesRelation + include Enumerable + + attr_reader :status + attr_accessor :offset_value, :limit_value + + delegate :last, :[], :to_s, :reverse, to: :to_a + + ALL_BATCHES_LIMIT = 100_000_000 # When no limit value it defaults to "all batches" + + def initialize(queue_adapter:, status: nil) + @queue_adapter = queue_adapter + @status = status + + set_defaults + end + + def offset(offset) + clone_with offset_value: offset + end + + def limit(limit) + clone_with limit_value: limit + end + + def each(&block) + batches.each(&block) + end + + def reload + @count = @batches = nil + self + end + + def count + if loaded? + to_a.length + else + query_count + end + end + + def empty? + count == 0 + end + + alias length count + alias size count + + private + def set_defaults + self.offset_value = 0 + self.limit_value = ALL_BATCHES_LIMIT + end + + def batches + @batches ||= @queue_adapter.batches(status: status, offset: offset_value, limit: limit_value).collect do |attributes| + MissionControl::Jobs::Batch.new(queue_adapter: @queue_adapter, **attributes) + end + end + + # The adapter count ignores pagination and can be +Float::INFINITY+ when + # internally limited, so clamp it into the offset/limit window here. + def query_count + @count ||= begin + count = [ @queue_adapter.batches_count(status: status) - offset_value, 0 ].max + limit_value_provided? ? [ count, limit_value ].min : count + end + end + + def limit_value_provided? + limit_value.present? && limit_value != ALL_BATCHES_LIMIT + end + + def loaded? + !@batches.nil? + end + + def clone_with(**properties) + dup.reload.tap do |relation| + properties.each do |key, value| + relation.send("#{key}=", value) + end + end + end +end diff --git a/app/views/mission_control/jobs/batches/_batch.html.erb b/app/views/mission_control/jobs/batches/_batch.html.erb new file mode 100644 index 00000000..2e141663 --- /dev/null +++ b/app/views/mission_control/jobs/batches/_batch.html.erb @@ -0,0 +1,21 @@ + + + <%= link_to batch.id, application_batch_path(@application, batch.id) %> + + + <% if batch.description.present? %> + <%= batch.description %> + <% else %> + No description + <% end %> + + <%= render "mission_control/jobs/batches/status", batch: batch %> + <%= number_to_percentage batch.progress_percentage, precision: 0 %> + + <%= batch.total_jobs %> + <% if batch.failed_jobs > 0 %> + (<%= batch.failed_jobs %> failed) + <% end %> + +
<%= batch.enqueued_at ? bidirectional_time_distance_in_words_with_title(batch.enqueued_at) : "Not started" %>
+ diff --git a/app/views/mission_control/jobs/batches/_general_information.html.erb b/app/views/mission_control/jobs/batches/_general_information.html.erb new file mode 100644 index 00000000..33e72d30 --- /dev/null +++ b/app/views/mission_control/jobs/batches/_general_information.html.erb @@ -0,0 +1,37 @@ + + + + + + + + + + + <% if batch.finished_at.present? %> + + + + + <% end %> + <% if batch.metadata.present? %> + + + + + <% end %> + +
Progress + " value="<%= batch.progress_percentage %>" max="100"><%= number_to_percentage batch.progress_percentage, precision: 0 %> + <% job_counts = batch.job_counts %> + <% job_counts.each.with_index do |(status, count), index| %> + <% label = status == :finished ? "completed" : status.to_s.humanize.downcase %> + <% if count > 0 %> + <%= link_to "#{count} #{label}", application_batch_path(@application, batch.id, jobs_status: status), class: ("has-text-danger" if status == :failed) %> + <% else %> + <%= count %> <%= label %> + <% end %> + <%= "," unless index == job_counts.size - 1 %> + <% end %> + of <%= batch.total_jobs %> total +
Enqueued<%= batch.enqueued_at ? bidirectional_time_distance_in_words_with_title(batch.enqueued_at) : "Not started" %>
Finished<%= bidirectional_time_distance_in_words_with_title(batch.finished_at) %>
Metadata
<%= batch.metadata %>
diff --git a/app/views/mission_control/jobs/batches/_status.html.erb b/app/views/mission_control/jobs/batches/_status.html.erb new file mode 100644 index 00000000..581bcdf8 --- /dev/null +++ b/app/views/mission_control/jobs/batches/_status.html.erb @@ -0,0 +1 @@ +<%= batch.status %> diff --git a/app/views/mission_control/jobs/batches/_title.html.erb b/app/views/mission_control/jobs/batches/_title.html.erb new file mode 100644 index 00000000..c3816cf4 --- /dev/null +++ b/app/views/mission_control/jobs/batches/_title.html.erb @@ -0,0 +1,13 @@ +

+
+
+ Batch <%= batch.id %> + <% if batch.description.present? %> + <%= batch.description %> + <% end %> +
+
+ <%= render "mission_control/jobs/batches/status", batch: batch %> +
+
+

diff --git a/app/views/mission_control/jobs/batches/index.html.erb b/app/views/mission_control/jobs/batches/index.html.erb new file mode 100644 index 00000000..8d89b82c --- /dev/null +++ b/app/views/mission_control/jobs/batches/index.html.erb @@ -0,0 +1,37 @@ +<% navigation(title: "Batches", section: :batches) %> + +<% if @batches_page.empty? && batches_status.nil? %> + <%= blank_status_notice "There are no batches" %> +<% else %> +
+ +
+ + <% if @batches_page.empty? %> + <%= blank_status_notice "No #{batches_status} batches found" %> + <% else %> + + + + + + + + + + + + + + <%= render partial: "mission_control/jobs/batches/batch", collection: @batches_page.records %> + +
DescriptionStatusProgressJobsEnqueued
+ + <%= render "mission_control/jobs/shared/pagination_toolbar", page: @batches_page, filter_param: batches_filter_param %> + <% end %> +<% end %> diff --git a/app/views/mission_control/jobs/batches/show.html.erb b/app/views/mission_control/jobs/batches/show.html.erb new file mode 100644 index 00000000..9acd244f --- /dev/null +++ b/app/views/mission_control/jobs/batches/show.html.erb @@ -0,0 +1,18 @@ +<% navigation(title: "Batch #{@batch.id}", section: :batches) %> + +<%= render "mission_control/jobs/batches/title", batch: @batch %> +<%= render "mission_control/jobs/batches/general_information", batch: @batch %> + +<% if @jobs_page.empty? %> + <%= blank_status_notice "No #{jobs_status.dasherize} jobs found for this batch" %> +<% else %> +

<%= pluralize @jobs_page.total_count, "#{jobs_status.dasherize} job" %>

+ + <% if jobs_status == "pending" %> + <%= render "mission_control/jobs/shared/jobs", jobs: @jobs_page.records %> + <% else %> + <%= render "mission_control/jobs/jobs/jobs_page", jobs_page: @jobs_page %> + <% end %> + + <%= render "mission_control/jobs/shared/pagination_toolbar", page: @jobs_page, filter_param: jobs_filter_param.merge(jobs_status: jobs_status) %> +<% end %> diff --git a/config/routes.rb b/config/routes.rb index e5b6fd88..7f683aba 100644 --- a/config/routes.rb +++ b/config/routes.rb @@ -21,6 +21,7 @@ resources :workers, only: [ :index, :show ] resources :recurring_tasks, only: [ :index, :show, :update ] + resources :batches, only: [ :index, :show ] end # Allow referencing urls without providing an application_id. It will default to the first one. diff --git a/lib/active_job/jobs_relation.rb b/lib/active_job/jobs_relation.rb index 0d71e5e4..e78987c0 100644 --- a/lib/active_job/jobs_relation.rb +++ b/lib/active_job/jobs_relation.rb @@ -25,7 +25,7 @@ class ActiveJob::JobsRelation STATUSES = %i[ pending failed in_progress blocked scheduled finished ] FILTERS = %i[ queue_name job_class_name ] - PROPERTIES = %i[ queue_name status offset_value limit_value job_class_name worker_id recurring_task_id finished_at ] + PROPERTIES = %i[ queue_name status offset_value limit_value job_class_name worker_id recurring_task_id batch_id finished_at ] attr_reader *PROPERTIES, :default_page_size delegate :last, :[], :reverse, to: :to_a @@ -51,13 +51,15 @@ def initialize(queue_adapter: ActiveJob::Base.queue_adapter, default_page_size: # * :queue_name - To only include the jobs in the provided queue. # * :worker_id - To only include the jobs processed by the provided worker. # * :recurring_task_id - To only include the jobs corresponding to runs of a recurring task. + # * :batch_id - To only include the jobs belonging to a given batch. # * :finished_at - (Range) To only include the jobs finished between the provided range - def where(job_class_name: nil, queue_name: nil, worker_id: nil, recurring_task_id: nil, finished_at: nil) + def where(job_class_name: nil, queue_name: nil, worker_id: nil, recurring_task_id: nil, batch_id: nil, finished_at: nil) # Remove nil arguments to avoid overriding parameters when concatenating +where+ clauses arguments = { job_class_name: job_class_name, queue_name: queue_name&.to_s, worker_id: worker_id, recurring_task_id: recurring_task_id, + batch_id: batch_id, finished_at: finished_at }.compact diff --git a/lib/active_job/queue_adapters/solid_queue_ext.rb b/lib/active_job/queue_adapters/solid_queue_ext.rb index 392d834a..e777bd75 100644 --- a/lib/active_job/queue_adapters/solid_queue_ext.rb +++ b/lib/active_job/queue_adapters/solid_queue_ext.rb @@ -1,6 +1,6 @@ module ActiveJob::QueueAdapters::SolidQueueExt include MissionControl::Jobs::Adapter - include RecurringTasks, Workers + include RecurringTasks, Workers, Batches def queues queues = SolidQueue::Queue.all @@ -73,8 +73,8 @@ def dispatch_job(job, jobs_relation) dispatch_immediately find_solid_queue_job!(job.job_id, jobs_relation) end - def find_job(job_id, *) - if job = SolidQueue::Job.where(active_job_id: job_id).order(:id).last + def find_job(job_id, jobs_relation) + if job = find_solid_queue_job(job_id, jobs_relation) deserialize_and_proxy_solid_queue_job job end end @@ -180,7 +180,7 @@ def retry_all attr_reader :jobs_relation delegate :queue_name, :limit_value, :limit_value_provided?, :offset_value, :job_class_name, - :default_page_size, :worker_id, :recurring_task_id, :finished_at, to: :jobs_relation + :default_page_size, :worker_id, :recurring_task_id, :batch_id, :finished_at, to: :jobs_relation def executions execution_class_by_status @@ -189,6 +189,7 @@ def executions .then { |executions| filter_executions_by_class(executions) } .then { |executions| filter_executions_by_process_id(executions) } .then { |executions| filter_executions_by_task_key(executions) } + .then { |executions| filter_executions_by_batch(executions) } .then { |executions| limit(executions) } .then { |executions| offset(executions) } end @@ -197,6 +198,7 @@ def finished_jobs SolidQueue::Job.finished .then { |jobs| filter_jobs_by_queue(jobs) } .then { |jobs| filter_jobs_by_class(jobs) } + .then { |jobs| filter_jobs_by_batch(jobs) } .then { |jobs| filter_jobs_by_finished_at(jobs) } .then { |jobs| limit(jobs) } .then { |jobs| offset(jobs) } @@ -216,7 +218,7 @@ def order_executions(executions) end def matches_relation_filters?(job) - matches_status?(job) && matches_queue_name?(job) + matches_status?(job) && matches_queue_name?(job) && matches_batch?(job) end def direct_count @@ -275,6 +277,14 @@ def filter_executions_by_task_key(executions) recurring_task_id.present? ? executions.where(task_key: recurring_task_id) : executions end + def filter_executions_by_batch(executions) + batch_id.present? ? executions.where(job: { batch_id: batch_id }) : executions + end + + def filter_jobs_by_batch(jobs) + batch_id.present? ? jobs.where(batch_id: batch_id) : jobs + end + def filter_jobs_by_class(jobs) job_class_name.present? ? jobs.where(class_name: job_class_name) : jobs end @@ -299,6 +309,10 @@ def matches_queue_name?(job) queue_name.blank? || job.queue_name == queue_name end + def matches_batch?(job) + batch_id.blank? || job.batch_id.to_s == batch_id.to_s + end + def solid_queue_status STATUS_MAP[jobs_relation.status].to_s.inquiry end diff --git a/lib/active_job/queue_adapters/solid_queue_ext/batches.rb b/lib/active_job/queue_adapters/solid_queue_ext/batches.rb new file mode 100644 index 00000000..b13fa4bf --- /dev/null +++ b/lib/active_job/queue_adapters/solid_queue_ext/batches.rb @@ -0,0 +1,162 @@ +module ActiveJob::QueueAdapters::SolidQueueExt::Batches + BATCHES_LIMIT = 100 + BATCH_COLUMNS = %w[ id description metadata total_jobs completed_jobs failed_jobs enqueued_at finished_at failed_at ] + BATCH_EXECUTION_COLUMNS = %w[ job_id batch_id ] + JOB_COLUMNS = %w[ id batch_id ] + JOB_EXECUTION_CLASS_NAMES = { + pending: "SolidQueue::ReadyExecution", + failed: "SolidQueue::FailedExecution", + in_progress: "SolidQueue::ClaimedExecution", + blocked: "SolidQueue::BlockedExecution", + scheduled: "SolidQueue::ScheduledExecution" + } + + def supports_batches? + solid_queue_batch_models_available? && + SolidQueue::Batch.method_defined?(:status) && + solid_queue_batch_schema_available? + rescue ActiveRecord::ActiveRecordError, NameError + false + end + + def batches(status: nil, offset: 0, limit: BATCHES_LIMIT) + batches_relation(status).merge(batches_scope(status)).order(id: :desc).offset(offset).limit(limit).collect do |batch| + batch_attributes_from_solid_queue_batch(batch) + end + end + + def batches_count(status: nil) + count_limit = MissionControl::Jobs.internal_query_count_limit + 1 + limited_count = batches_scope(status).limit(count_limit).count + (limited_count == count_limit) ? Float::INFINITY : limited_count + end + + def find_batch(batch_id) + if batch = solid_queue_batches.find_by(id: batch_id) + batch_attributes_from_solid_queue_batch(batch) + end + end + + private + def batches_scope(status) + case status + when :finished then SolidQueue::Batch.finished + when :unfinished then SolidQueue::Batch.unfinished + when :failed then SolidQueue::Batch.failed + else SolidQueue::Batch.all + end + end + + # Finished/failed batches already store counters on the row. Skip live job-count + # subqueries there so listing millions of historical batches stays cheap. + def batches_relation(status) + case status + when :finished, :failed then SolidQueue::Batch.all + else solid_queue_batches + end + end + + def batch_attributes_from_solid_queue_batch(batch) + job_counts = job_counts_from_solid_queue_batch(batch) + unfinished_jobs = batch_count(batch, :unfinished) + + if batch.finished_at.present? + completed_jobs = batch[:completed_jobs] + job_counts = job_counts.transform_values { 0 }.merge(failed: batch[:failed_jobs]) + unfinished_jobs = 0 + else + completed_jobs = batch.total_jobs - unfinished_jobs - job_counts[:failed] + end + + { + id: batch.id, + description: batch.description, + status: batch.status, + total_jobs: batch.total_jobs, + completed_jobs: completed_jobs, + failed_jobs: job_counts[:failed], + pending_jobs: job_counts[:pending], + in_progress_jobs: job_counts[:in_progress], + blocked_jobs: job_counts[:blocked], + scheduled_jobs: job_counts[:scheduled], + progress_percentage: progress_percentage(batch.total_jobs, unfinished_jobs), + metadata: batch.metadata, + enqueued_at: batch.enqueued_at, + finished_at: batch.finished_at + } + end + + def solid_queue_batch_models_available? + %i[ Batch BatchExecution ].all? { |name| SolidQueue.const_defined?(name, false) } + end + + def solid_queue_batch_schema_available? + model_has_columns?(SolidQueue::Batch, BATCH_COLUMNS) && + model_has_columns?(SolidQueue::BatchExecution, BATCH_EXECUTION_COLUMNS) && + model_has_columns?(SolidQueue::Job, JOB_COLUMNS) && + job_execution_models.all? { |model| model_has_columns?(model, [ "job_id" ]) } + end + + def model_has_columns?(model, columns) + model.table_exists? && columns.all? { |column| model.column_names.include?(column) } + end + + def solid_queue_batches + selects = [ "#{SolidQueue::Batch.quoted_table_name}.*", unfinished_jobs_count_select ] + selects.concat JOB_EXECUTION_CLASS_NAMES.map { |status, class_name| job_count_select(status, class_name.constantize) } + SolidQueue::Batch.select(*selects) + end + + # CASE keeps Postgres from evaluating the correlated COUNT for finished rows + # when an "all" listing mixes finished and unfinished batches. + def unfinished_jobs_count_select + batch_executions_table = SolidQueue::BatchExecution.quoted_table_name + batches_table = SolidQueue::Batch.quoted_table_name + batch_id = quote_column_name("batch_id") + id = quote_column_name("id") + finished_at = quote_column_name("finished_at") + live_count = "SELECT COUNT(*) FROM #{batch_executions_table} WHERE #{batch_executions_table}.#{batch_id} = #{batches_table}.#{id}" + + "(CASE WHEN #{batches_table}.#{finished_at} IS NOT NULL THEN 0 ELSE (#{live_count}) END) AS #{quote_column_name(batch_count_attribute(:unfinished))}" + end + + def job_count_select(status, execution_model) + executions_table = execution_model.quoted_table_name + jobs_table = SolidQueue::Job.quoted_table_name + batches_table = SolidQueue::Batch.quoted_table_name + job_id = quote_column_name("job_id") + batch_id = quote_column_name("batch_id") + id = quote_column_name("id") + finished_at = quote_column_name("finished_at") + finished_value = status == :failed ? "#{batches_table}.#{quote_column_name("failed_jobs")}" : "0" + live_count = "SELECT COUNT(*) FROM #{executions_table} INNER JOIN #{jobs_table} ON #{jobs_table}.#{id} = #{executions_table}.#{job_id} WHERE #{jobs_table}.#{batch_id} = #{batches_table}.#{id}" + + "(CASE WHEN #{batches_table}.#{finished_at} IS NOT NULL THEN #{finished_value} ELSE (#{live_count}) END) AS #{quote_column_name(batch_count_attribute(status))}" + end + + def job_counts_from_solid_queue_batch(batch) + JOB_EXECUTION_CLASS_NAMES.keys.to_h { |status| [ status, batch_count(batch, status) ] } + end + + def batch_count(batch, status) + batch[batch_count_attribute(status)].to_i + end + + def batch_count_attribute(status) + "mission_control_#{status}_jobs" + end + + def progress_percentage(total_jobs, unfinished_jobs) + return 0 if total_jobs == 0 + + ((total_jobs - unfinished_jobs) * 100.0 / total_jobs).round(2) + end + + def job_execution_models + JOB_EXECUTION_CLASS_NAMES.values.map(&:constantize) + end + + def quote_column_name(name) + SolidQueue::Batch.connection.quote_column_name(name) + end +end diff --git a/lib/mission_control/jobs/adapter.rb b/lib/mission_control/jobs/adapter.rb index 3db8c497..7e6cbd9e 100644 --- a/lib/mission_control/jobs/adapter.rb +++ b/lib/mission_control/jobs/adapter.rb @@ -80,6 +80,51 @@ def find_worker(worker_id) end + def supports_batches? + false + end + + # Returns an array with the requested page of batches, newest first, honoring + # +offset+ and +limit+. +status+ narrows the list to +:finished+, +:unfinished+, + # or +:failed+ batches when present. Each batch is represented as a hash with + # these attributes: + # { + # id: 123, + # description: "Nightly imports", + # status: :enqueued, + # total_jobs: 100, + # completed_jobs: 60, + # failed_jobs: 2, + # pending_jobs: 20, + # in_progress_jobs: 10, + # blocked_jobs: 5, + # scheduled_jobs: 3, + # progress_percentage: 62.0, + # metadata: { user_id: 123 }, + # enqueued_at: Fri, 26 Jan 2024 20:31:09.652174000 UTC +00:00, + # finished_at: nil + # } + def batches(status: nil, offset: 0, limit: nil) + if supports_batches? + raise_incompatible_adapter_error_from :batches + end + end + + # Returns the total number of batches, narrowed by +status+ when present + def batches_count(status: nil) + if supports_batches? + raise_incompatible_adapter_error_from :batches_count + end + end + + # Returns a batch represented by a hash as indicated above + def find_batch(batch_id) + if supports_batches? + raise_incompatible_adapter_error_from :find_batch + end + end + + # Returns an array with the list of queues. Each queue is represented as a hash # with these attributes: # { diff --git a/lib/mission_control/jobs/server.rb b/lib/mission_control/jobs/server.rb index 970d5f5a..d2281f52 100644 --- a/lib/mission_control/jobs/server.rb +++ b/lib/mission_control/jobs/server.rb @@ -2,7 +2,7 @@ class MissionControl::Jobs::Server include MissionControl::Jobs::IdentifiedByName - include Serializable, RecurringTasks, Workers + include Serializable, RecurringTasks, Workers, Batches attr_reader :name, :queue_adapter, :application, :backtrace_cleaner diff --git a/lib/mission_control/jobs/server/batches.rb b/lib/mission_control/jobs/server/batches.rb new file mode 100644 index 00000000..f95b9730 --- /dev/null +++ b/lib/mission_control/jobs/server/batches.rb @@ -0,0 +1,13 @@ +module MissionControl::Jobs::Server::Batches + def batches(status: nil) + MissionControl::Jobs::BatchesRelation.new(queue_adapter: queue_adapter, status: status) + end + + def find_batch(batch_id) + if batch = queue_adapter.find_batch(batch_id) + MissionControl::Jobs::Batch.new(queue_adapter: queue_adapter, **batch) + else + raise MissionControl::Jobs::Errors::ResourceNotFound, "Batch with id '#{batch_id}' not found" + end + end +end diff --git a/mission_control-jobs.gemspec b/mission_control-jobs.gemspec index fc01761c..dd1c8c0f 100644 --- a/mission_control-jobs.gemspec +++ b/mission_control-jobs.gemspec @@ -35,7 +35,7 @@ Gem::Specification.new do |spec| spec.add_dependency "irb", "~> 1.13" spec.add_development_dependency "resque" - spec.add_development_dependency "solid_queue", "~> 1.0.1" + spec.add_development_dependency "solid_queue", "~> 1.0" spec.add_development_dependency "selenium-webdriver" spec.add_development_dependency "resque-pause" spec.add_development_dependency "mocha" diff --git a/test/active_job/queue_adapters/solid_queue_test.rb b/test/active_job/queue_adapters/solid_queue_test.rb index c5d1036e..40840a46 100644 --- a/test/active_job/queue_adapters/solid_queue_test.rb +++ b/test/active_job/queue_adapters/solid_queue_test.rb @@ -8,7 +8,96 @@ class ActiveJob::QueueAdapters::SolidQueueTest < ActiveSupport::TestCase SolidQueue.logger = ActiveSupport::Logger.new(nil) end + test "supports batches only when the Solid Queue batch API is available" do + assert ActiveJob::Base.queue_adapter.supports_batches? + + SolidQueue.stubs(:const_defined?).with(:Batch, false).returns(false) + + assert_not ActiveJob::Base.queue_adapter.supports_batches? + end + + test "supports batches only when the Solid Queue batch schema is installed" do + SolidQueue::Batch.stubs(:table_exists?).returns(false) + + assert_not ActiveJob::Base.queue_adapter.supports_batches? + end + + test "find a job by id applies the batch filter" do + batch_1, job_1 = create_batch + _batch_2, job_2 = create_batch + + batch_jobs = ActiveJob.jobs.where(batch_id: batch_1.id) + + assert_equal job_1.job_id, batch_jobs.find_by_id(job_1.job_id).job_id + assert_nil batch_jobs.find_by_id(job_2.job_id) + end + + test "loads batches and their job counts in one query" do + 3.times { create_batch } + + queries = capture_select_queries do + batches = ActiveJob::Base.queue_adapter.batches + assert_equal [ 1, 1, 1 ], batches.pluck(:pending_jobs) + end + + assert_equal 1, queries.size, queries.join("\n\n") + end + + test "lists finished batches without live job-count subqueries" do + batch, = create_batch + batch.update!(finished_at: Time.current, completed_jobs: 1, failed_jobs: 0) + + queries = capture_select_queries do + batches = ActiveJob::Base.queue_adapter.batches(status: :finished) + assert_equal [ batch.id ], batches.pluck(:id) + assert_equal [ 0 ], batches.pluck(:pending_jobs) + assert_equal [ 1 ], batches.pluck(:completed_jobs) + end + + assert_equal 1, queries.size, queries.join("\n\n") + assert_no_match(/solid_queue_batch_executions|solid_queue_ready_executions|solid_queue_failed_executions/, queries.first) + end + + test "caps batches count like job counts" do + 3.times { create_batch } + + original_limit = MissionControl::Jobs.internal_query_count_limit + MissionControl::Jobs.internal_query_count_limit = 2 + + assert_equal Float::INFINITY, ActiveJob::Base.queue_adapter.batches_count + assert_equal Float::INFINITY, ActiveJob::Base.queue_adapter.batches_count(status: :unfinished) + ensure + MissionControl::Jobs.internal_query_count_limit = original_limit + end + + test "returns an exact batches count below the internal limit" do + 2.times { create_batch } + + original_limit = MissionControl::Jobs.internal_query_count_limit + MissionControl::Jobs.internal_query_count_limit = 5 + + assert_equal 2, ActiveJob::Base.queue_adapter.batches_count(status: :unfinished) + ensure + MissionControl::Jobs.internal_query_count_limit = original_limit + end + private + def create_batch + job = nil + batch = SolidQueue::Batch.enqueue { job = DummyJob.perform_later } + [ batch, job ] + end + + def capture_select_queries + queries = [] + subscriber = lambda do |*, payload| + queries << payload[:sql] if payload[:sql].match?(/\ASELECT\b/i) + end + + ActiveSupport::Notifications.subscribed(subscriber, "sql.active_record") { yield } + queries + end + def queue_adapter :solid_queue end diff --git a/test/controllers/batches_controller_test.rb b/test/controllers/batches_controller_test.rb new file mode 100644 index 00000000..a27721dc --- /dev/null +++ b/test/controllers/batches_controller_test.rb @@ -0,0 +1,251 @@ +require "test_helper" + +class MissionControl::Jobs::BatchesControllerTest < ActionDispatch::IntegrationTest + # File-local job so no shared job class gets its queue adapter mutated + class BatchedJob < ActiveJob::Base + self.queue_adapter = :solid_queue + queue_as :default + + def perform(*); end + end + + test "get batch list" do + create_batch(description: "Nightly imports") + + get mission_control_jobs.application_batches_url(@application) + assert_response :ok + + assert_select "tr.batch", 1 + assert_select "td", "Nightly imports" + assert_select "span.tag", "enqueued" + assert_select "li.is-active a", "Unfinished" + end + + test "batch list is paginated" do + 12.times { |i| create_batch(description: "Batch #{i}") } + + get mission_control_jobs.application_batches_url(@application) + assert_response :ok + assert_select "tr.batch", 10 + assert_select "nav[aria-label=pagination]" + + get mission_control_jobs.application_batches_url(@application, page: 2) + assert_response :ok + assert_select "tr.batch", 2 + end + + test "batch list filters by finished, unfinished, and failed" do + create_batch(description: "Still going") + finished = create_batch(description: "All done") + failed = create_batch(description: "Went wrong") + SolidQueue::Job.where(batch_id: finished.id).each { |job| finish(job) } + SolidQueue::Job.where(batch_id: failed.id).each { |job| fail_job(job, RuntimeError.new("boom")) } + + get mission_control_jobs.application_batches_url(@application, batches_status: "finished") + assert_response :ok + assert_select "tr.batch", 2 # a failed batch has also finished + + get mission_control_jobs.application_batches_url(@application, batches_status: "unfinished") + assert_response :ok + assert_select "tr.batch", 1 + assert_select "td", "Still going" + + get mission_control_jobs.application_batches_url(@application, batches_status: "failed") + assert_response :ok + assert_select "tr.batch", 1 + assert_select "td", "Went wrong" + + get mission_control_jobs.application_batches_url(@application) + assert_select "tr.batch", 1 + assert_select "td", "Still going" + + get mission_control_jobs.application_batches_url(@application, batches_status: "all") + assert_select "tr.batch", 3 + end + + test "batch list pagination preserves the status filter" do + 12.times { create_batch } + + get mission_control_jobs.application_batches_url(@application, batches_status: "unfinished") + assert_response :ok + assert_select "tr.batch", 10 + assert_select "nav[aria-label=pagination] a[href*=?]", "batches_status=unfinished" + end + + test "batch list shows an empty notice for a status filter without matches" do + create_batch(description: "Still going") + + get mission_control_jobs.application_batches_url(@application, batches_status: "finished") + assert_response :ok + assert_select "tr.batch", 0 + assert_select "div", text: "No finished batches found" + end + + test "batch list defaults to unfinished and shows an empty notice when there are none" do + get mission_control_jobs.application_batches_url(@application) + assert_response :ok + + assert_select "tr.batch", 0 + assert_select "li.is-active a", "Unfinished" + assert_select "div", text: "No unfinished batches found" + end + + test "batch list shows an empty notice when there are no batches on the all tab" do + get mission_control_jobs.application_batches_url(@application, batches_status: "all") + assert_response :ok + + assert_select "tr.batch", 0 + assert_select "div", text: "There are no batches" + end + + test "get batch details and pending job list" do + batch = create_batch(description: "Nightly imports", jobs: 2) + + get mission_control_jobs.application_batch_url(@application, batch.id) + assert_response :ok + + assert_select "h1", /Batch #{batch.id}/ + assert_select "h2", "2 pending jobs" + assert_select "tr.job", 2 + end + + test "batch details expose each unfinished job status separately" do + batch = create_batch_with_unfinished_statuses + + get mission_control_jobs.application_batch_url(@application, batch.id) + assert_response :ok + + assert_select "td a", "1 pending" + assert_select "td a", "1 in progress" + assert_select "td a", "1 blocked" + assert_select "td a", "1 scheduled" + + %w[ pending in_progress blocked scheduled ].each do |status| + get mission_control_jobs.application_batch_url(@application, batch.id, jobs_status: status) + assert_response :ok + + assert_select "h2", "1 #{status.dasherize} job" + assert_select "tr.job", 1 + end + end + + test "scheduled-only batch shows its scheduled jobs by default" do + batch = create_batch(jobs: 2, wait: 1.hour) + + get mission_control_jobs.application_batch_url(@application, batch.id) + assert_response :ok + + assert_select "h2", "2 scheduled jobs" + assert_select "tr.job", 2 + end + + test "batch progress reflects finished jobs" do + batch = create_batch(jobs: 4) + finish SolidQueue::Job.where(batch_id: batch.id).order(:id).first + + get mission_control_jobs.application_batch_url(@application, batch.id) + assert_response :ok + + assert_select "td a", "1 completed" + assert_select "td a", "3 pending" + assert_select "td", /of 4 total/ + end + + test "failed batch shows its failed jobs with error details" do + batch = create_batch(description: "Doomed", jobs: 2) + jobs = SolidQueue::Job.where(batch_id: batch.id).order(:id).to_a + fail_job jobs.first, RuntimeError.new("boom went the job") + finish jobs.second + + get mission_control_jobs.application_batch_url(@application, batch.id) + assert_response :ok + + assert_select "span.tag", "failed" + assert_select "h2", "1 failed job" + assert_select "tr.job", 1 + assert_select "td", /boom went the job/ + assert_select "td a", "1 failed" + end + + test "jobs_status param switches the job list" do + batch = create_batch(jobs: 3) + finish SolidQueue::Job.where(batch_id: batch.id).order(:id).first + + get mission_control_jobs.application_batch_url(@application, batch.id, jobs_status: :finished) + assert_response :ok + + assert_select "h2", "1 finished job" + assert_select "tr.job", 1 + end + + test "pagination preserves the selected jobs status" do + batch = create_batch(jobs: 3) + SolidQueue::Job.where(batch_id: batch.id).find_each { |job| finish(job) } + + stub_const(MissionControl::Jobs::Page, :DEFAULT_PAGE_SIZE, 2) do + get mission_control_jobs.application_batch_url(@application, batch.id, jobs_status: :finished) + assert_response :ok + + next_page_url = css_select("a.pagination-next").find { |link| link.text == "Next page" }["href"] + assert_includes next_page_url, "jobs_status=finished" + + get next_page_url + assert_response :ok + + assert_select "h2", "3 finished jobs" + assert_select "tr.job", 1 + end + end + + test "redirect to batches list when batch doesn't exist" do + get mission_control_jobs.application_batch_url(@application, 987654) + assert_redirected_to mission_control_jobs.application_batches_url(@application) + + follow_redirect! + + assert_select "article.is-danger", /Batch with id '987654' not found/ + end + + private + def create_batch(description: nil, jobs: 1, wait: nil) + SolidQueue::Batch.enqueue(description: description) do + jobs.times do |i| + wait ? BatchedJob.set(wait: wait).perform_later(i) : BatchedJob.perform_later(i) + end + end + end + + def create_batch_with_unfinished_statuses + batch = SolidQueue::Batch.enqueue do + 3.times { |i| BatchedJob.perform_later(i) } + BatchedJob.set(wait: 1.hour).perform_later(3) + end + + claimed_job_id, blocked_job_id = SolidQueue::ReadyExecution.where(job_id: batch.jobs.ids).order(:job_id).limit(2).pluck(:job_id) + SolidQueue::ReadyExecution.where(job_id: [ claimed_job_id, blocked_job_id ]).delete_all + + SolidQueue::ClaimedExecution.insert_all!([ { job_id: claimed_job_id, process_id: nil, created_at: Time.current } ]) + SolidQueue::Job.where(id: blocked_job_id).update_all(concurrency_key: "batch-key") + SolidQueue::BlockedExecution.insert_all!([ { + job_id: blocked_job_id, + queue_name: "default", + priority: 0, + concurrency_key: "batch-key", + expires_at: 1.hour.from_now, + created_at: Time.current + } ]) + + batch + end + + # Mirror the real worker flow: the execution is claimed away before the job resolves + def finish(job) + SolidQueue::ReadyExecution.where(job_id: job.id).destroy_all + job.finished! + end + + def fail_job(job, error) + SolidQueue::ReadyExecution.where(job_id: job.id).destroy_all + job.failed_with(error) + end +end diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb index 85194b6a..f9a71dab 100644 --- a/test/dummy/db/queue_schema.rb +++ b/test/dummy/db/queue_schema.rb @@ -37,7 +37,9 @@ t.string "concurrency_key" t.datetime "created_at", null: false t.datetime "updated_at", null: false + t.bigint "batch_id" t.index [ "active_job_id" ], name: "index_solid_queue_jobs_on_active_job_id" + t.index [ "batch_id" ], name: "index_solid_queue_jobs_on_batch_id" t.index [ "class_name" ], name: "index_solid_queue_jobs_on_class_name" t.index [ "finished_at" ], name: "index_solid_queue_jobs_on_finished_at" t.index [ "queue_name", "finished_at" ], name: "index_solid_queue_jobs_for_filtering" @@ -120,6 +122,35 @@ t.index [ "key" ], name: "index_solid_queue_semaphores_on_key", unique: true end + create_table "solid_queue_batches", force: :cascade do |t| + t.string "active_job_batch_id" + t.string "description" + t.text "on_finish" + t.text "on_success" + t.text "on_failure" + t.text "metadata" + t.integer "total_jobs", default: 0, null: false + t.integer "completed_jobs", default: 0, null: false + t.integer "failed_jobs", default: 0, null: false + t.datetime "enqueued_at" + t.datetime "finished_at" + t.datetime "failed_at" + t.datetime "created_at", null: false + t.datetime "updated_at", null: false + t.index [ "active_job_batch_id" ], name: "index_solid_queue_batches_on_active_job_batch_id", unique: true + t.index [ "finished_at" ], name: "index_solid_queue_batches_on_finished_at" + end + + create_table "solid_queue_batch_executions", force: :cascade do |t| + t.bigint "job_id", null: false + t.bigint "batch_id", null: false + t.datetime "created_at", null: false + t.index [ "job_id" ], name: "index_solid_queue_batch_executions_on_job_id", unique: true + t.index [ "batch_id" ], name: "index_solid_queue_batch_executions_on_batch_id" + end + + add_foreign_key "solid_queue_batch_executions", "solid_queue_batches", column: "batch_id", on_delete: :cascade + add_foreign_key "solid_queue_batch_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade diff --git a/test/test_helper.rb b/test/test_helper.rb index 9c298b1a..f05ce625 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -61,6 +61,7 @@ def delete_solid_queue_data SolidQueue::Job.find_each(&:destroy) SolidQueue::Process.find_each(&:destroy) SolidQueue::RecurringTask.find_each(&:destroy) + SolidQueue::Batch.find_each(&:destroy) end def root_resque_redis