Merge pull request #294 from ZeusWPI/chore/switch-to-solid-queue

Switch from delayed job to solid queue

Gemfile

@@ -31,8 +31,8 @@ gem 'github_api'
 gem 'sentry-rails'
 gem 'sentry-ruby'
 
-# Use delayed job for executing jobs in the background
-gem 'delayed_job_active_record'
+# Use solid queue for executing jobs in the background
+gem 'solid_queue'
 
 # Set CORS headers
 gem 'rack-cors'

Gemfile.lock

@@ -103,15 +103,12 @@ GEM
     debug (1.9.2)
       irb (~> 1.10)
       reline (>= 0.3.8)
-    delayed_job (4.1.12)
-      activesupport (>= 3.0, < 8.0)
-    delayed_job_active_record (4.1.10)
-      activerecord (>= 3.0, < 8.0)
-      delayed_job (>= 3.0, < 5)
     descendants_tracker (0.0.4)
       thread_safe (~> 0.3, >= 0.3.1)
     drb (2.2.1)
     erubi (1.13.0)
+    et-orbi (1.2.11)
+      tzinfo
     faraday (1.10.3)
       faraday-em_http (~> 1.0)
       faraday-em_synchrony (~> 1.0)
@@ -135,6 +132,9 @@ GEM
     faraday-patron (1.0.0)
     faraday-rack (1.0.0)
     faraday-retry (1.0.3)
+    fugit (1.11.1)
+      et-orbi (~> 1, >= 1.2.11)
+      raabro (~> 1.4)
     github_api (0.19.0)
       addressable (~> 2.4)
       descendants_tracker (~> 0.0.4)
@@ -202,6 +202,7 @@ GEM
     public_suffix (5.0.4)
     puma (6.4.3)
       nio4r (~> 2.0)
+    raabro (1.4.0)
     racc (1.8.1)
     rack (3.1.8)
     rack-cors (2.0.2)
@@ -285,6 +286,13 @@ GEM
     sentry-ruby (5.21.0)
       bigdecimal
       concurrent-ruby (~> 1.0, >= 1.0.2)
+    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)
     sprockets (4.2.1)
       concurrent-ruby (~> 1.0)
       rack (>= 2.2.4, < 4)
@@ -330,7 +338,6 @@ DEPENDENCIES
   capybara
   cssbundling-rails
   debug
-  delayed_job_active_record
   github_api
   jbuilder
   pg (~> 1.5)
@@ -343,6 +350,7 @@ DEPENDENCIES
   selenium-webdriver
   sentry-rails
   sentry-ruby
+  solid_queue
   sprockets-rails
   tzinfo-data
   unicorn (~> 6.1)

app/jobs/repository_reprocess_job.rb

@@ -0,0 +1,9 @@
+class RepositoryReprocessJob < ApplicationJob
+  queue_as :default
+
+  def perform(repository)
+    OrganisationMember.create_all_from_organisation(repository.organisation)
+    repository.pull_or_clone
+    repository.process_commits
+  end
+end

app/models/organisation_member.rb

@@ -14,12 +14,13 @@ class OrganisationMember < ApplicationRecord
   validates :github_name, uniqueness: { scope: :organisation }
 
   def self.create_all_from_organisation(organisation)
-    OrganisationMember.where(organisation:).destroy_all
-    # rubocop:disable Rails/FindEach
-    # This is not a rails structure
-    Rails.application.config.github.orgs.members.all(organisation).each do |member|
-      OrganisationMember.create(organisation:, github_name: member['login'])
+    members = Rails.application.config.github.orgs.members.all(organisation)
+
+    OrganisationMember.transaction do
+      OrganisationMember.where(organisation:).destroy_all
+      members.each do |member|
+        OrganisationMember.create(organisation:, github_name: member['login'])
+      end
     end
-    # rubocop:enable Rails/FindEach
   end
 end

app/models/repository.rb

@@ -23,7 +23,7 @@ class Repository < ApplicationRecord
   validate :not_filtered
 
   before_create :set_path
-  after_save :reprocess_delayed
+  after_save :reprocess_in_background
 
   def self.create_or_update_from_github_api(organisation, name, github_url, clone_url)
     Repository.find_or_initialize_by(organisation:, name:) do |r|
@@ -32,34 +32,6 @@ def self.create_or_update_from_github_api(organisation, name, github_url, clone_
     end.save
   end
 
-  private
-
-  def set_path
-    self.path = Rails.root.join("repos/#{organisation.parameterize}_#{name.parameterize}")
-  end
-
-  def not_filtered
-    errors.add(:name, 'filtered') if Rails.application.config.repo_name_deny_list.include?(name)
-  end
-
-  def reprocess_delayed
-    delay.reprocess
-  end
-
-  def reprocess
-    OrganisationMember.create_all_from_organisation(organisation)
-    pull_or_clone
-    process_commits
-  end
-
-  def process_commits
-    rugged_repo.tap do |r|
-      walker = Rugged::Walker.new(r)
-      r.branches.each { |b| walker.push b.target_id }
-      walker.each { |c| Commit.from_rugged(c, self) }
-    end
-  end
-
   def pull_or_clone
     if Dir.exist?(path)
       logger.info("Fetching #{name} and resetting to FETCH_HEAD")
@@ -72,6 +44,28 @@ def pull_or_clone
     end
   end
 
+  def process_commits
+    rugged_repo.tap do |r|
+      walker = Rugged::Walker.new(r)
+      r.branches.each { |b| walker.push b.target_id }
+      walker.each { |c| Commit.from_rugged(c, self) }
+    end
+  end
+
+  private
+
+  def set_path
+    self.path = Rails.root.join("repos/#{organisation.parameterize}_#{name.parameterize}")
+  end
+
+  def not_filtered
+    errors.add(:name, 'filtered') if Rails.application.config.repo_name_deny_list.include?(name)
+  end
+
+  def reprocess_in_background
+    RepositoryReprocessJob.perform_later(self)
+  end
+
   def rugged_repo
     Rugged::Repository.new(path)
   end

bin/jobs

@@ -0,0 +1,6 @@
+#!/usr/bin/env ruby
+
+require_relative "../config/environment"
+require "solid_queue/cli"
+
+SolidQueue::Cli.start(ARGV)

config/application.rb

@@ -23,7 +23,7 @@ class Application < Rails::Application
     config.load_defaults 7.2
     config.active_support.cache_format_version = 7.1
     config.autoload_lib(ignore: %w[assets tasks])
-    config.active_job.queue_adapter = :delayed_job
+    config.active_job.queue_adapter = :solid_queue
     config.organisations = %w[ZeusWPI 12urenloop]
     config.repo_name_deny_list = %w[
       Bestuurstaakjes

config/environments/production.rb

@@ -62,8 +62,6 @@
   # Use a different cache store in production.
   # config.cache_store = :mem_cache_store
 
-  # Use a real queuing backend for Active Job (and separate queues per environment).
-  # config.active_job.queue_adapter = :resque
   # config.active_job.queue_name_prefix = "gamification2_production"
 
   # Enable locale fallbacks for I18n (makes lookups for any locale fall back to

config/puma.rb

@@ -28,6 +28,7 @@
 
 # Allow puma to be restarted by `bin/rails restart` command.
 plugin :tmp_restart
+plugin :solid_queue
 
 # Only use a pidfile when requested
 pidfile ENV['PIDFILE'] if ENV['PIDFILE']

config/queue.yml

@@ -0,0 +1,18 @@
+default: &default
+  dispatchers:
+    - polling_interval: 1
+      batch_size: 500
+  workers:
+    - queues: "*"
+      threads: 1
+      processes: <%= ENV.fetch("JOB_CONCURRENCY", 1) %>
+      polling_interval: 0.1
+
+development:
+  <<: *default
+
+test:
+  <<: *default
+
+production:
+  <<: *default

config/recurring.yml

@@ -0,0 +1,10 @@
+# production:
+#   periodic_cleanup:
+#     class: CleanSoftDeletedRecordsJob
+#     queue: background
+#     args: [ 1000, { batch_size: 500 } ]
+#     schedule: every hour
+#   periodic_command:
+#     command: "SoftDeletedRecord.due.delete_all"
+#     priority: 2
+#     schedule: at 5am every day

db/migrate/20241109090736_remove_delayed_job.rb

@@ -0,0 +1,5 @@
+class RemoveDelayedJob < ActiveRecord::Migration[7.2]
+  def change
+    drop_table :delayed_jobs
+  end
+end

db/migrate/20241109091236_add_solid_queue_tables.rb

@@ -0,0 +1,131 @@
+class AddSolidQueueTables < ActiveRecord::Migration[7.2]
+  def change
+    create_table 'solid_queue_blocked_executions', force: :cascade do |t|
+      t.bigint 'job_id', null: false
+      t.string 'queue_name', null: false
+      t.integer 'priority', default: 0, null: false
+      t.string 'concurrency_key', null: false
+      t.datetime 'expires_at', null: false
+      t.datetime 'created_at', null: false
+      t.index %w[concurrency_key priority job_id], name: 'index_solid_queue_blocked_executions_for_release'
+      t.index %w[expires_at concurrency_key], name: 'index_solid_queue_blocked_executions_for_maintenance'
+      t.index ['job_id'], name: 'index_solid_queue_blocked_executions_on_job_id', unique: true
+    end
+
+    create_table 'solid_queue_claimed_executions', force: :cascade do |t|
+      t.bigint 'job_id', null: false
+      t.bigint 'process_id'
+      t.datetime 'created_at', null: false
+      t.index ['job_id'], name: 'index_solid_queue_claimed_executions_on_job_id', unique: true
+      t.index %w[process_id job_id], name: 'index_solid_queue_claimed_executions_on_process_id_and_job_id'
+    end
+
+    create_table 'solid_queue_failed_executions', force: :cascade do |t|
+      t.bigint 'job_id', null: false
+      t.text 'error'
+      t.datetime 'created_at', null: false
+      t.index ['job_id'], name: 'index_solid_queue_failed_executions_on_job_id', unique: true
+    end
+
+    create_table 'solid_queue_jobs', force: :cascade do |t|
+      t.string 'queue_name', null: false
+      t.string 'class_name', null: false
+      t.text 'arguments'
+      t.integer 'priority', default: 0, null: false
+      t.string 'active_job_id'
+      t.datetime 'scheduled_at'
+      t.datetime 'finished_at'
+      t.string 'concurrency_key'
+      t.datetime 'created_at', null: false
+      t.datetime 'updated_at', null: false
+      t.index ['active_job_id'], name: 'index_solid_queue_jobs_on_active_job_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 %w[queue_name finished_at], name: 'index_solid_queue_jobs_for_filtering'
+      t.index %w[scheduled_at finished_at], name: 'index_solid_queue_jobs_for_alerting'
+    end
+
+    create_table 'solid_queue_pauses', force: :cascade do |t|
+      t.string 'queue_name', null: false
+      t.datetime 'created_at', null: false
+      t.index ['queue_name'], name: 'index_solid_queue_pauses_on_queue_name', unique: true
+    end
+
+    create_table 'solid_queue_processes', force: :cascade do |t|
+      t.string 'kind', null: false
+      t.datetime 'last_heartbeat_at', null: false
+      t.bigint 'supervisor_id'
+      t.integer 'pid', null: false
+      t.string 'hostname'
+      t.text 'metadata'
+      t.datetime 'created_at', null: false
+      t.string 'name', null: false
+      t.index ['last_heartbeat_at'], name: 'index_solid_queue_processes_on_last_heartbeat_at'
+      t.index %w[name supervisor_id], name: 'index_solid_queue_processes_on_name_and_supervisor_id', unique: true
+      t.index ['supervisor_id'], name: 'index_solid_queue_processes_on_supervisor_id'
+    end
+
+    create_table 'solid_queue_ready_executions', force: :cascade do |t|
+      t.bigint 'job_id', null: false
+      t.string 'queue_name', null: false
+      t.integer 'priority', default: 0, null: false
+      t.datetime 'created_at', null: false
+      t.index ['job_id'], name: 'index_solid_queue_ready_executions_on_job_id', unique: true
+      t.index %w[priority job_id], name: 'index_solid_queue_poll_all'
+      t.index %w[queue_name priority job_id], name: 'index_solid_queue_poll_by_queue'
+    end
+
+    create_table 'solid_queue_recurring_executions', force: :cascade do |t|
+      t.bigint 'job_id', null: false
+      t.string 'task_key', null: false
+      t.datetime 'run_at', null: false
+      t.datetime 'created_at', null: false
+      t.index ['job_id'], name: 'index_solid_queue_recurring_executions_on_job_id', unique: true
+      t.index %w[task_key run_at], name: 'index_solid_queue_recurring_executions_on_task_key_and_run_at', unique: true
+    end
+
+    create_table 'solid_queue_recurring_tasks', force: :cascade do |t|
+      t.string 'key', null: false
+      t.string 'schedule', null: false
+      t.string 'command', limit: 2048
+      t.string 'class_name'
+      t.text 'arguments'
+      t.string 'queue_name'
+      t.integer 'priority', default: 0
+      t.boolean 'static', default: true, null: false
+      t.text 'description'
+      t.datetime 'created_at', null: false
+      t.datetime 'updated_at', null: false
+      t.index ['key'], name: 'index_solid_queue_recurring_tasks_on_key', unique: true
+      t.index ['static'], name: 'index_solid_queue_recurring_tasks_on_static'
+    end
+
+    create_table 'solid_queue_scheduled_executions', force: :cascade do |t|
+      t.bigint 'job_id', null: false
+      t.string 'queue_name', null: false
+      t.integer 'priority', default: 0, null: false
+      t.datetime 'scheduled_at', null: false
+      t.datetime 'created_at', null: false
+      t.index ['job_id'], name: 'index_solid_queue_scheduled_executions_on_job_id', unique: true
+      t.index %w[scheduled_at priority job_id], name: 'index_solid_queue_dispatch_all'
+    end
+
+    create_table 'solid_queue_semaphores', force: :cascade do |t|
+      t.string 'key', null: false
+      t.integer 'value', default: 1, null: false
+      t.datetime 'expires_at', null: false
+      t.datetime 'created_at', null: false
+      t.datetime 'updated_at', null: false
+      t.index ['expires_at'], name: 'index_solid_queue_semaphores_on_expires_at'
+      t.index %w[key value], name: 'index_solid_queue_semaphores_on_key_and_value'
+      t.index ['key'], name: 'index_solid_queue_semaphores_on_key', unique: true
+    end
+
+    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
+    add_foreign_key 'solid_queue_ready_executions', 'solid_queue_jobs', column: 'job_id', on_delete: :cascade
+    add_foreign_key 'solid_queue_recurring_executions', 'solid_queue_jobs', column: 'job_id', on_delete: :cascade
+    add_foreign_key 'solid_queue_scheduled_executions', 'solid_queue_jobs', column: 'job_id', on_delete: :cascade
+  end
+end