Add coalesce to deduplicate pending jobs

README.md

@@ -28,6 +28,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Errors when enqueuing](#errors-when-enqueuing)
 - [Concurrency controls](#concurrency-controls)
   - [Performance considerations](#performance-considerations)
+- [Coalescing](#coalescing)
 - [Failed jobs and retries](#failed-jobs-and-retries)
   - [Error reporting on jobs](#error-reporting-on-jobs)
   - [Jobs interrupted by non-graceful process death](#jobs-interrupted-by-non-graceful-process-death)
@@ -647,6 +648,28 @@ In addition, mixing concurrency controls with **bulk enqueuing** (Active Job's `
 
 When jobs that have concurrency controls and `on_conflict: :discard` are enqueued in bulk, the ones that fail to be enqueued and are discarded would have `successfully_enqueued` set to `false`. The total count of jobs enqueued returned by `perform_all_later` will exclude these jobs as expected.
 
+## Coalescing
+
+Coalescing makes jobs with the same key share one pending job. The first job enqueued with a key is created, and later ones with the same key, including scheduled ones, are discarded until that job starts. Once a worker claims it, the key is free again, so a job enqueued while it runs creates the next pending job.
+
+```ruby
+class ReindexProductJob < ApplicationJob
+  coalesce key: ->(product_id) { product_id }
+
+  def perform(product_id)
+    Product.find(product_id).reindex!
+  end
+end
+```
+
+`key` works like the key in [concurrency controls](#concurrency-controls), and it's scoped to the job class, and to the batch for jobs in a [batch](#batch-jobs). The pending job runs with the first job's arguments, so pass IDs and load fresh data in `perform`.
+
+After enqueuing, `coalesce_status` is `:created` or `:discarded`, and `provider_job_id` is the ID of the pending job either way. Retries are never coalesced.
+
+Coalescing limits how many jobs wait, not how many run, so two jobs with the same key can run at the same time. Combine it with `limits_concurrency` to prevent that.
+
+Existing installations need a new table for coalescing. See [UPGRADING.md](UPGRADING.md).
+
 ## Failed jobs and retries
 
 Solid Queue doesn't include any automatic retry mechanism, it [relies on Active Job for this](https://edgeguides.rubyonrails.org/active_job_basics.html#retrying-or-discarding-failed-jobs). Jobs that fail will be kept in the system, and a _failed execution_ (a record in the `solid_queue_failed_executions` table) will be created for these. The job will stay there until manually discarded or re-enqueued. You can do this in a console as:

UPGRADING.md

@@ -1,3 +1,13 @@
+# Upgrading to version 1.8.x
+This version adds [coalescing](README.md#coalescing), which needs a new table. Fresh installs get it with the base schema; existing installations need to copy the migration that adds it and run it:
+
+```bash
+bin/rails solid_queue:update
+bin/rails db:migrate
+```
+
+Until then, everything works as before, and enqueuing a job that uses `coalesce` raises `SolidQueue::CoalesceKey::PendingMigrations`.
+
 # Upgrading to version 1.7.2
 Jobs with [concurrency controls](README.md#concurrency-controls) now remember which semaphore they got their slot from, in their serialized job data, and give the slot back only to that semaphore. Before, a job that ran past its `duration` gave its slot back to the semaphore a later job had created after the job's own one expired, so one more job than the limit allowed could run at the same time, and keep doing so until the jobs waiting for that key ran out.
 

app/models/solid_queue/coalesce_key.rb

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class CoalesceKey < Record
+    class PendingMigrations < StandardError
+      def initialize(message = "The coalescing schema hasn't been installed yet. Run `bin/rails solid_queue:update` to copy the pending migrations to your application, and then `bin/rails db:migrate` to run them")
+        super
+      end
+    end
+
+    belongs_to :job, optional: true
+
+    class << self
+      def migrated?
+        @migrated ||= table_exists?
+      end
+
+      # Locked until the transaction commits, so the key's job can't be claimed in the meantime
+      def admit(key)
+        insert_unless_held(key)
+
+        # Retry if the key's job was claimed between the insert and the lock
+        lock.find_by(key: key) || admit(key)
+      end
+
+      # Locks the keys in order, like bulk admission, so the two can't deadlock
+      def release_all(job_ids)
+        if migrated? && (ids = where(job_id: job_ids).order(:key).lock.ids).any?
+          where(id: ids).delete_all
+        end
+      end
+
+      private
+        # PostgreSQL aborts the current transaction on a duplicate key, so skip
+        # duplicates instead of raising
+        def insert_unless_held(key)
+          if supports_insert_conflict_target?
+            insert({ key: key }, unique_by: :key)
+          else
+            insert({ key: key })
+          end
+        end
+    end
+  end
+end

app/models/solid_queue/job.rb

@@ -13,7 +13,7 @@ def self.for(job)
       end
     end
 
-    include Executable, Clearable, Recurrable, Batchable
+    include Executable, Clearable, Recurrable, Batchable, Coalescing
 
     serialize :arguments, coder: JSON
 
@@ -26,10 +26,12 @@ def enqueue_all(active_jobs)
           job.scheduled_at ||= Time.current
           job.batch_id = current_batch_id || job.batch_id
         end
-        active_jobs_by_job_id = active_jobs.index_by(&:job_id)
+
+        coalesced_jobs, uncoalesced_jobs = active_jobs.partition(&:coalesced?)
+        active_jobs_by_job_id = uncoalesced_jobs.index_by(&:job_id)
 
         transaction do
-          jobs = create_all_from_active_jobs(active_jobs)
+          jobs = create_all_from_active_jobs(uncoalesced_jobs)
           prepare_all_for_execution(jobs).tap do |enqueued_jobs|
             enqueued_jobs.each do |enqueued_job|
               active_jobs_by_job_id[enqueued_job.active_job_id].provider_job_id = enqueued_job.id
@@ -38,15 +40,22 @@ def enqueue_all(active_jobs)
           end
         end
 
+        # In key order, so that concurrent bulk enqueues lock the keys they share in the same order
+        coalesced_jobs.sort_by(&:coalesce_key).each { |active_job| enqueue_coalesced(active_job) }
+
         active_jobs.count(&:successfully_enqueued?)
       end
 
       def enqueue(active_job, scheduled_at: Time.current)
         active_job.scheduled_at = scheduled_at
 
-        create_from_active_job(active_job).tap do |enqueued_job|
-          active_job.provider_job_id = enqueued_job.id if enqueued_job.persisted?
-          active_job.successfully_enqueued = enqueued_job.persisted?
+        if active_job.coalesced?
+          enqueue_coalesced(active_job)
+        else
+          create_from_active_job(active_job).tap do |enqueued_job|
+            active_job.provider_job_id = enqueued_job.id if enqueued_job.persisted?
+            active_job.successfully_enqueued = enqueued_job.persisted?
+          end
         end
       end
 

app/models/solid_queue/job/coalescing.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class Job
+    module Coalescing
+      extend ActiveSupport::Concern
+
+      class_methods do
+        def enqueue_coalesced(active_job)
+          raise CoalesceKey::PendingMigrations unless CoalesceKey.migrated?
+
+          transaction do
+            coalesce_key = CoalesceKey.admit(active_job.coalesce_key)
+
+            if coalesce_key.job_id
+              discard_into(coalesce_key.job_id, active_job)
+            else
+              create_for(coalesce_key, active_job)
+            end
+          end
+        rescue ActiveRecord::ActiveRecordError => e
+          enqueue_error = EnqueueError.new("#{e.class.name}: #{e.message}").tap do |error|
+            error.set_backtrace e.backtrace
+          end
+          raise enqueue_error
+        end
+
+        private
+          def discard_into(job_id, active_job)
+            active_job.provider_job_id = job_id
+            active_job.successfully_enqueued = false
+            active_job.coalesce_status = :discarded
+            nil
+          end
+
+          def create_for(coalesce_key, active_job)
+            create_from_active_job(active_job).tap do |job|
+              if job.persisted?
+                coalesce_key.update!(job: job)
+                active_job.provider_job_id = job.id
+                active_job.coalesce_status = :created
+              else
+                coalesce_key.destroy!
+              end
+
+              active_job.successfully_enqueued = job.persisted?
+            end
+          end
+      end
+    end
+  end
+end

app/models/solid_queue/ready_execution.rb

@@ -40,6 +40,9 @@ def lock_candidates(executions, process_id)
           SolidQueue::ClaimedExecution.claiming(executions.map(&:job_id), process_id) do |claimed|
             ids_to_delete = executions.index_by(&:job_id).values_at(*claimed.map(&:job_id)).map(&:id)
             where(id: ids_to_delete).delete_all
+
+            # A job that starts frees its key, so the next call queues another job
+            CoalesceKey.release_all(claimed.map(&:job_id))
           end
         end
 

lib/active_job/coalescing.rb

@@ -0,0 +1,43 @@
+# frozen_string_literal: true
+
+module ActiveJob
+  module Coalescing
+    extend ActiveSupport::Concern
+
+    included do
+      class_attribute :coalesce_key, instance_accessor: false
+    end
+
+    attr_accessor :coalesce_status
+
+    class_methods do
+      def coalesce(key:)
+        self.coalesce_key = key
+      end
+    end
+
+    def coalesce_key
+      if self.class.coalesce_key
+        param = compute_concurrency_parameter(self.class.coalesce_key)
+
+        case param
+        when ActiveRecord::Base
+          [ coalesce_group, param.class.name, param.id ]
+        else
+          [ coalesce_group, param ]
+        end.compact.join("/")
+      end
+    end
+
+    # Retries continue work that has already started, so they're never coalesced
+    def coalesced?
+      coalesce_key.present? && executions.zero?
+    end
+
+    private
+      # A batch waits for its own jobs, so a call in a batch only joins a job in the same batch
+      def coalesce_group
+        [ ("Batch/#{batch_id}" if batch_id), self.class.name ].compact.join("/")
+      end
+  end
+end

lib/generators/solid_queue/install/templates/db/queue_schema.rb

@@ -149,10 +149,19 @@
     t.index [ "batch_id" ], name: "index_solid_queue_batch_executions_on_batch_id"
   end
 
+  create_table "solid_queue_coalesce_keys", force: :cascade do |t|
+    t.string "key", null: false
+    t.bigint "job_id"
+    t.datetime "created_at", null: false
+    t.index [ "job_id" ], name: "index_solid_queue_coalesce_keys_on_job_id", unique: true
+    t.index [ "key" ], name: "index_solid_queue_coalesce_keys_on_key", unique: true
+  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_coalesce_keys", "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

lib/generators/solid_queue/update/templates/db/add_coalesce_keys_to_solid_queue.rb

@@ -0,0 +1,15 @@
+class AddCoalesceKeysToSolidQueue < ActiveRecord::Migration[7.1]
+  def change
+    # Fresh installs create this with the base schema, so skip it if it
+    # already exists
+    create_table :solid_queue_coalesce_keys, if_not_exists: true do |t|
+      t.string :key, null: false
+      t.bigint :job_id
+      t.datetime :created_at, null: false
+
+      t.index :key, unique: true
+      t.index :job_id, unique: true
+      t.foreign_key :solid_queue_jobs, column: :job_id, on_delete: :cascade
+    end
+  end
+end

lib/solid_queue/engine.rb

@@ -41,6 +41,7 @@ class Engine < ::Rails::Engine
     initializer "solid_queue.active_job.extensions" do
       ActiveSupport.on_load :active_job do
         include ActiveJob::ConcurrencyControls
+        include ActiveJob::Coalescing
 
         ActiveSupport.on_load :active_record do
           ActiveJob::Base.include ActiveJob::BatchId

test/dummy/app/jobs/coalesced_job.rb

@@ -0,0 +1,7 @@
+class CoalescedJob < ApplicationJob
+  coalesce key: ->(name) { name }
+
+  def perform(name)
+    JobBuffer.add(name)
+  end
+end

test/dummy/db/queue_schema.rb

@@ -161,10 +161,19 @@
     t.index ["batch_id"], name: "index_solid_queue_batch_executions_on_batch_id"
   end
 
+  create_table "solid_queue_coalesce_keys", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
+    t.string "key", null: false
+    t.bigint "job_id"
+    t.datetime "created_at", null: false
+    t.index ["job_id"], name: "index_solid_queue_coalesce_keys_on_job_id", unique: true
+    t.index ["key"], name: "index_solid_queue_coalesce_keys_on_key", unique: true
+  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_coalesce_keys", "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