Merge pull request #2257 from coreinfrastructure/solid_queue

Add solid_queue as active_jobs backend

Author: david-a-wheeler

Gemfile

@@ -100,6 +100,7 @@ gem 'redcarpet', '~> 3.5' # Process markdown in form textareas (justifications)
 gem 'sassc-rails' # compiles .scss (css replacement) drop-in replacement for sass-rails
 gem 'scout_apm' # Monitor for memory leaks
 gem 'secure_headers', '~> 7' # Add hardening measures to HTTP headers
+gem 'solid_queue', '~> 1.1' # ActiveJob database backend
 # WARNING!!!!
 # CHECK DEPLOYMENT FIRST IF YOU UPDATE sprockets-rails.
 # The gem sprockets-rails version 3.4.1 (from 3.2.2) caused a regression

Gemfile.lock

@@ -146,6 +146,8 @@ GEM
       execjs
       multi_json (>= 1.3)
       rake
+    et-orbi (1.2.11)
+      tzinfo
     execjs (2.10.0)
     faraday (2.12.2)
       faraday-net_http (>= 2.0, < 3.5)
@@ -169,6 +171,9 @@ GEM
       nokogiri (>= 1.11.3)
       railties (>= 4.2)
     forwardable (1.3.3)
+    fugit (1.11.1)
+      et-orbi (~> 1, >= 1.2.11)
+      raabro (~> 1.4)
     get_process_mem (1.0.0)
       bigdecimal (>= 2.0)
       ffi (~> 1.0)
@@ -359,6 +364,7 @@ GEM
       bigdecimal (>= 2.0)
       get_process_mem (>= 0.2)
       puma (>= 2.7)
+    raabro (1.4.0)
     racc (1.8.1)
     rack (2.2.10)
     rack-attack (6.7.0)
@@ -501,6 +507,13 @@ GEM
     snaky_hash (2.0.1)
       hashie
       version_gem (~> 1.1, >= 1.1.1)
+    solid_queue (1.1.2)
+      activejob (>= 7.1)
+      activerecord (>= 7.1)
+      concurrent-ruby (>= 1.3.1)
+      fugit (~> 1.11.0)
+      railties (>= 7.1)
+      thor (~> 1.3.1)
     spring (4.2.1)
     sprockets (4.2.1)
       concurrent-ruby (~> 1.0)
@@ -641,6 +654,7 @@ DEPENDENCIES
   selenium-webdriver
   sentry-rails
   sentry-ruby
+  solid_queue (~> 1.1)
   spring (~> 4.1)
   sprockets-rails (= 3.5.2)
   translation (= 1.41)

bin/jobs

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

config/environments/production.rb

@@ -239,5 +239,12 @@
     # Do nothing if it's unavailable (this happens if we didn't load the gem)
   end
   # rubocop:enable Lint/HandleExceptions
+
+  # Configure active_job to use solid_queue as its back end when in production.
+  config.active_job.queue_adapter = :solid_queue
+  # We intentionally use *one* database for both the jobs and everythihg else.
+  # This greatly simplifies backup, maintenance, etc. If you wanted the
+  # jobs to go into their own database, you'd need to do something like:
+  # config.solid_queue.connects_to = { database: { writing: :queue } }
 end
 # rubocop:enable Metrics/BlockLength

config/puma.rb

@@ -54,6 +54,11 @@
 # Allow puma to be restarted by `rails restart` command.
 plugin :tmp_restart
 
+# Use solid_queue as the ActiveJob backend. This stores jobs in a database,
+# so scheduled jobs will happen even if the system crashes.
+# Set the environment variable if you want it activated (e.g., in production)
+plugin :solid_queue if ENV['SOLID_QUEUE_IN_PUMA'] # || Rails.env.development?
+
 # Use puma_worker_killer to occasionally restart.
 # This is a band-aid to counter memory growth.
 # There's a performance hit (restart time + cache loss), but it

config/queue.yml

@@ -0,0 +1,18 @@
+default: &default
+  dispatchers:
+    - polling_interval: 1
+      batch_size: 100
+  workers:
+    - queues: "*"
+      threads: 2
+      processes: <%= ENV.fetch("JOB_CONCURRENCY", 1) %>
+      polling_interval: 0.5
+
+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/20241227211821_solid_queue.rb

@@ -0,0 +1,134 @@
+# frozen_string_literal: true
+# Generated by solid_queue.
+
+class SolidQueue < ActiveRecord::Migration[8.0]
+  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 [ "concurrency_key", "priority", "job_id" ], name: "index_solid_queue_blocked_executions_for_release"
+      t.index [ "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 [ "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 [ "queue_name", "finished_at" ], name: "index_solid_queue_jobs_for_filtering"
+      t.index [ "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 [ "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 [ "priority", "job_id" ], name: "index_solid_queue_poll_all"
+      t.index [ "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 [ "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 [ "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 [ "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