chore: solid_queue v1.6.0 → v1.7.0 — update Mongoid models for upstream changes

[github-actions]

solid_queue v1.7.0 — Upstream Model Changes

Comparing rails/solid_queue models between v1.6.0 → v1.7.0.
Each section shows what changed upstream alongside our current Mongoid model for context.
Review each diff and decide if the corresponding Mongoid model needs updating.

Summary

File	Upstream Change
claimed_execution.rb	🔄 Modified
failed_execution.rb	🔄 Modified
job.rb	🔄 Modified
job/executable.rb	🔄 Modified
batch.rb	🆕 Added in upstream
batch/callbacks.rb	🆕 Added in upstream
batch/clearable.rb	🆕 Added in upstream
batch/status.rb	🆕 Added in upstream
batch/sweepable.rb	🆕 Added in upstream
batch_execution.rb	🆕 Added in upstream
failed_execution/batchable.rb	🆕 Added in upstream
job/batchable.rb	🆕 Added in upstream

Review Checklist

• lib/solid_queue_mongoid/models/claimed_execution.rb — review upstream changes
• lib/solid_queue_mongoid/models/failed_execution.rb — review upstream changes
• lib/solid_queue_mongoid/models/job.rb — review upstream changes
• lib/solid_queue_mongoid/models/job/executable.rb — review upstream changes
• lib/solid_queue_mongoid/models/batch.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/batch/callbacks.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/batch/clearable.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/batch/status.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/batch/sweepable.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/batch_execution.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/failed_execution/batchable.rb — new upstream file, consider adding
• lib/solid_queue_mongoid/models/job/batchable.rb — new upstream file, consider adding

---

Detailed Diffs

---

batch.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class Batch < Record
    class AlreadyFinished < StandardError; end

    class PendingMigrations < StandardError
      def initialize(message = "The batches 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

    include Callbacks, Status
    include Clearable, Sweepable

    has_many :jobs
    has_many :batch_executions, dependent: :destroy

    store :metadata, coder: JSON

    # Join-free so update_all keeps this condition in the completion update's own WHERE
    scope :without_executions, -> { where.not(id: BatchExecution.select(:batch_id)) }

    # Provider-agnostic batch identifier, analogous to jobs.active_job_id.
    before_create :set_active_job_batch_id
    after_commit :start, on: :create, unless: -> { ActiveRecord.respond_to?(:after_all_transactions_commit) }

    class << self
      # The batches schema ships as an optional migration in Solid Queue 1.x
      # and becomes part of the base schema in 2.0. Until the app has run the
      # migration, jobs enqueue without any batch bookkeeping and batches
      # themselves can't be used.
      def migrated?
        @migrated ||= table_exists? && BatchExecution.table_exists? && Job.column_names.include?("batch_id")
      end

      def enqueue(description: nil, on_success: nil, on_failure: nil, on_finish: nil, metadata: nil, **extra_metadata, &block)
        raise PendingMigrations unless migrated?

        new.tap do |batch|
          batch.assign_attributes(description:, on_success:, on_failure:, on_finish:, metadata: (metadata || {}).merge(extra_metadata))
          batch.enqueue(&block)
        end
      end

      def current_batch_id
        ActiveSupport::IsolatedExecutionState[:current_batch_id]
      end

      def wrap_in_batch_context(batch_id)
        previous_batch_id = current_batch_id.presence
        ActiveSupport::IsolatedExecutionState[:current_batch_id] = batch_id
        yield
      ensure
        ActiveSupport::IsolatedExecutionState[:current_batch_id] = previous_batch_id
      end
    end

    def enqueue(&block)
      # Fast-fail for the common case. create_all_from_jobs atomically guards
      # concurrent additions when it creates their tracking rows.
      if finished?
        raise AlreadyFinished, "Can't enqueue an already finished batch"
      end

      transaction do
        save! if new_record?

        self.class.wrap_in_batch_context(id) { block&.call(self) }

        if ActiveRecord.respond_to?(:after_all_transactions_commit)
          ActiveRecord.after_all_transactions_commit { start }
        end
      end
    end

    def metadata
      (super || {}).with_indifferent_access
    end

    def start
      mark_as_enqueued

      # Refresh enqueued_at after marking as enqueued, and let a batch that started
      # with no jobs finish right away
      reload
      finish
    end

    def finish
      return if finished? || !enqueued?
      return if batch_executions.exists?

      transaction do
        updated = Batch.where(id: id).unfinished.enqueued.without_executions.update_all(finished_at: Time.current)
        finalize if updated > 0
      end
    end

    private
      def set_active_job_batch_id
        self.active_job_batch_id ||= SecureRandom.uuid
      end

      def mark_as_enqueued
        Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current)
      end

      def finalize
        reload

        # PostgreSQL can let a blocked CAS win from a stale NOT EXISTS snapshot:
        # after a lock wait, READ COMMITTED re-checks the target row's conditions
        # against the latest data but keeps the original snapshot for subqueries.
        # Re-check in a new statement, which gets a fresh snapshot while this
        # transaction's row lock keeps adders out, since they increment before
        # inserting their executions. MySQL doesn't need this: it reads DML
        # subqueries from the latest committed data, so its CAS can't win wrongly.
        raise ActiveRecord::Rollback if batch_executions.exists?

        SolidQueue.instrument(:finish_batch, batch_id: id) do |payload|
          failed_jobs = jobs.failed.count
          failed_at = Time.current if failed_jobs > 0
          completed_jobs = total_jobs - failed_jobs

          update_columns(failed_jobs:, failed_at:, completed_jobs:)
          enqueue_callback_jobs

          payload.merge!(total_jobs:, failed_jobs:, completed_jobs:)
        end
      end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/batch.rb`

No local equivalent exists for this file.

---

batch/callbacks.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class Batch
    module Callbacks
      extend ActiveSupport::Concern

      included do
        %w[ finish success failure ].each do |callback_type|
          serialize "on_#{callback_type}", coder: JSON

          define_method("on_#{callback_type}=") do |callback|
            super serialize_callback(callback)
          end
        end
      end

      private
        def serialize_callback(value)
          if value.present?
            active_job = value.is_a?(ActiveJob::Base) ? value : value.new
            # We can pick up batch ids from context, but callbacks should never be considered a part of the batch
            active_job.batch_id = nil
            active_job.serialize
          end
        end

        def enqueue_callback_jobs
          if failed? then enqueue_callback_job(:on_failure)
          else
            enqueue_callback_job(:on_success)
          end

          enqueue_callback_job(:on_finish)
        end

        def enqueue_callback_job(callback_name)
          if callback = send(callback_name)
            active_job = ActiveJob::Base.deserialize(callback)
            active_job.callback_batch_id = id
            # Bypass the job class's adapter so callbacks stay in Solid Queue and
            # their enqueue stays in this transaction, while honoring enqueue callbacks.
            active_job.run_callbacks(:enqueue) do
              Job.enqueue(active_job, scheduled_at: active_job.scheduled_at || Time.current)
            end
          end
        end
    end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/batch/callbacks.rb`

No local equivalent exists for this file.

---

batch/clearable.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class Batch
    module Clearable
      extend ActiveSupport::Concern

      included do
        scope :clearable, ->(finished_before: SolidQueue.clear_finished_jobs_after.ago) { succeeded.where(finished_at: ...finished_before) }
      end

      class_methods do
        def clear_finished_in_batches(batch_size: 500, finished_before: SolidQueue.clear_finished_jobs_after.ago, sleep_between_batches: 0)
          loop do
            records_deleted = clearable(finished_before: finished_before).limit(batch_size).delete_all
            sleep(sleep_between_batches) if sleep_between_batches > 0
            break if records_deleted == 0
          end
        end
      end
    end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/batch/clearable.rb`

No local equivalent exists for this file.

---

batch/status.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class Batch
    module Status
      extend ActiveSupport::Concern

      included do
        scope :finished, -> { where.not(finished_at: nil) }
        scope :succeeded, -> { finished.where(failed_at: nil) }
        scope :unfinished, -> { where(finished_at: nil) }
        scope :failed, -> { where.not(failed_at: nil) }
        scope :enqueued, -> { where.not(enqueued_at: nil) }
      end

      def status
        if finished?
          failed? ? :failed : :completed
        elsif enqueued?
          :enqueued
        else
          :pending
        end
      end

      def failed?
        failed_at.present?
      end

      def succeeded?
        finished? && !failed?
      end

      def finished?
        finished_at.present?
      end

      def enqueued?
        enqueued_at.present?
      end

      # Failed jobs no longer have tracking rows, so exclude them from the completed count.
      def completed_jobs
        finished? ? self[:completed_jobs] : [ total_jobs - pending_jobs - failed_jobs, 0 ].max
      end

      def failed_jobs
        finished? ? self[:failed_jobs] : jobs.failed.count
      end

      # Pending counts attempts, not logical jobs: while a retry is enqueued
      # and its previous attempt hasn't finished yet, both have tracking rows,
      # so the counts derived from it clamp at the logical totals.
      def pending_jobs
        finished? ? 0 : batch_executions.count
      end

      def progress_percentage
        return 0 if total_jobs == 0
        ([ total_jobs - pending_jobs, 0 ].max * 100.0 / total_jobs).round(2)
      end
    end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/batch/status.rb`

No local equivalent exists for this file.

---

batch/sweepable.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class Batch
    # Repairs batches that the regular completion detection can't finish on
    # its own: jobs removed via bulk discards, processes that crashed after
    # enqueueing jobs but before starting their batch, or completions whose
    # callback enqueueing failed and rolled back.
    module Sweepable
      extend ActiveSupport::Concern

      class_methods do
        def sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
          SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, stale_executions: 0, finished_batches: 0, started_batches: 0) do |payload|
            payload[:stale_executions] = sweep_stale_executions(batch_size:)
            payload[:finished_batches] = finish_stalled_batches(batch_size:)
            payload[:started_batches] = start_stalled_batches(stalled_for:, batch_size:)
          end
        end

        private
          # BatchExecution rows represent outstanding work. A row for a resolved
          # job violates that invariant, so remove it immediately; destroy's
          # after_commit callback retries the batch completion check.
          def sweep_stale_executions(batch_size:)
            swept = 0

            [ BatchExecution.with_finished_jobs, BatchExecution.with_failed_jobs ].each do |stale|
              stale.find_each(batch_size: batch_size) do |batch_execution|
                swept += 1
                batch_execution.destroy
              end
            end

            swept
          end

          # A started batch with no tracking rows left can finish
          def finish_stalled_batches(batch_size:)
            finished = 0

            unfinished.enqueued.without_executions.find_each(batch_size: batch_size) do |batch|
              finished += 1
              batch.finish
            end

            finished
          end

          # A batch that crashed between creation and start never got enqueued
          def start_stalled_batches(stalled_for:, batch_size:)
            started = 0

            unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
              started += 1
              batch.start
            end

            started
          end
      end
    end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/batch/sweepable.rb`

No local equivalent exists for this file.

---

batch_execution.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class BatchExecution < Execution
    self.assumable_attributes_from_job = [ :batch_id ]

    belongs_to :batch

    scope :with_finished_jobs, -> { joins(:job).merge(SolidQueue::Job.finished) }
    scope :with_failed_jobs, -> { joins(job: :failed_execution) }

    after_commit :finish_batch, on: :destroy

    class << self
      def create_all_from_jobs(jobs)
        jobs.select(&:batched?).group_by(&:batch_id).each do |batch_id, jobs_in_batch|
          # Update the counter first: inserting tracking rows takes a shared FK lock on
          # the batch row, then incrementing can deadlock concurrent MySQL adders.
          if attempt_to_update_total_jobs(batch_id, jobs_in_batch)
            super jobs_in_batch
          else
            raise Batch::AlreadyFinished, "Can't add jobs into an already finished batch"
          end
        end
      end

      private
        def attempt_to_update_total_jobs(batch_id, jobs)
          new_jobs_count = count_new_jobs_among(jobs)
          updated = SolidQueue::Batch.where(id: batch_id).unfinished.update_all([ "total_jobs = total_jobs + ?", new_jobs_count ])
          updated > 0
        end

        # A job that has executed before was already counted when it first joined
        # the batch: retries keep their active_job_id and batch across re-enqueues.
        # This might undercount jobs whose retries switch to another batch, but that
        # should be a rare enough case. The counter is used only for report/info, so
        # we favour simplicity here
        def count_new_jobs_among(jobs)
          jobs.reject { |job| job.arguments["executions"].to_i > 0 }.map(&:active_job_id).uniq.size
        end
    end

    private
      def finish_batch
        # Skip the serialized callback and metadata columns on this hot path
        if batch = Batch.select(:id, :finished_at, :enqueued_at).find_by(id: batch_id)
          batch.finish
        end
      end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/batch_execution.rb`

No local equivalent exists for this file.

---

claimed_execution.rb — 🔄 Modified upstream

📥 What changed in upstream (v1.6.0 → v1.7.0)

--- solid_queue@v1.6.0/claimed_execution.rb
+++ solid_queue@v1.7.0/claimed_execution.rb
@@ -48,6 +48,7 @@
           payload[:process_ids] = executions.map(&:process_id).uniq
           payload[:job_ids] = executions.map(&:job_id).uniq
           payload[:size] = executions.size
+          payload[:error] = error
         end
       end
     end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/claimed_execution.rb`

# frozen_string_literal: true

module SolidQueue
  class ClaimedExecution < Execution
    assumes_attributes_from_job # inherits queue_name and priority from job

    field :process_id, type: BSON::ObjectId

    belongs_to :process, class_name: "SolidQueue::Process", optional: true

    # Executions whose process_id references a process that no longer exists.
    scope :orphaned, lambda {
      existing_process_ids = SolidQueue::Process.all.pluck(:id)
      existing_process_ids.empty? ? all : where(:process_id.nin => existing_process_ids)
    }

    index({ process_id: 1 })

    Result = Struct.new(:success, :error) do
      def success?
        success
      end
    end

    class << self
      # Atomically creates ClaimedExecution records for the given job_ids and
      # yields the claimed set to the block (which deletes the ReadyExecutions).
      def claiming(job_ids, process_id, &block)
        job_data = Array(job_ids).map { |job_id| { job_id: job_id, process_id: process_id } }

        SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
          claimed = job_data.filter_map do |attrs|
            create!(attrs)
          rescue Mongoid::Errors::Validations, Mongo::Error::OperationFailure
            nil
          end

          block.call(claimed)

          payload[:size] = claimed.size
          payload[:claimed_job_ids] = claimed.map(&:job_id)
        end
      end

      def release_all
        SolidQueue.instrument(:release_many_claimed) do |payload|
          executions = all.to_a
          executions.each do |execution|
            execution.release
          rescue Mongoid::Errors::Validations, Mongo::Error::OperationFailure
            # If ReadyExecution already exists, that's fine
          end
          payload[:size] = executions.size
        end
      end

      def fail_all_with(error)
        executions = includes(:job).to_a
        return if executions.empty?

        SolidQueue.instrument(:fail_many_claimed) do |payload|
          executions.each do |execution|
            execution.failed_with(error)
            execution.unblock_next_job
          end
          payload[:process_ids] = executions.map(&:process_id).uniq
          payload[:job_ids] = executions.map(&:job_id).uniq
          payload[:size] = executions.size
        end
      end

      def discard_all_in_batches(*)
        raise Execution::UndiscardableError, "Can't discard jobs in progress"
      end

      def discard_all_from_jobs(*)
        raise Execution::UndiscardableError, "Can't discard jobs in progress"
      end
    end

    # Called by Pool thread — executes the job and marks it finished or failed.
    def perform
      result = execute

      if result.success?
        finished
      else
        failed_with(result.error)
        raise result.error
      end
    ensure
      unblock_next_job
    end

    # Release this execution back to ready (called by process deregister / prune).
    def release
      SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
        job.dispatch_bypassing_concurrency_limits
        destroy!
      end
    end

    def discard
      raise Execution::UndiscardableError, "Can't discard a job in progress"
    end

    def failed_with(error)
      Mongoid.transaction do
        job.failed_with(error)
        destroy!
      end
    end

    def unblock_next_job
      job.unblock_next_blocked_job
    end

    private

    def execute
      ActiveJob::Base.execute(job.arguments.merge("provider_job_id" => job.id.to_s))
      Result.new(true, nil)
    rescue Exception => e # rubocop:disable Lint/RescueException
      Result.new(false, e)
    end

    def finished
      Mongoid.transaction do
        job.finished!
        destroy!
      end
    end
  end
end

---

failed_execution.rb — 🔄 Modified upstream

📥 What changed in upstream (v1.6.0 → v1.7.0)

--- solid_queue@v1.6.0/failed_execution.rb
+++ solid_queue@v1.7.0/failed_execution.rb
@@ -2,7 +2,7 @@
 
 module SolidQueue
   class FailedExecution < Execution
-    include Dispatching
+    include Dispatching, Batchable
 
     serialize :error, coder: JSON

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/failed_execution.rb`

# frozen_string_literal: true

module SolidQueue
  class FailedExecution < Execution
    include Dispatching

    field :error, type: Hash # stores exception_class, message, backtrace

    attr_accessor :exception

    before_save :expand_error_details_from_exception, if: :exception

    index({ created_at: 1 })

    class << self
      def retry_all(jobs)
        SolidQueue.instrument(:retry_all, jobs_size: jobs.size) do |payload|
          job_ids = jobs.map(&:id)
          payload[:size] = dispatch_jobs(lock_all_from_jobs_ids(job_ids))
        end
      end

      private

      def lock_all_from_jobs_ids(job_ids)
        where(:job_id.in => job_ids).pluck(:job_id)
      end
    end

    def retry
      SolidQueue.instrument(:retry, job_id: job.id) do
        Mongoid.transaction do
          job.reset_execution_counters
          job.prepare_for_execution
          destroy!
        end
      end
    end

    # Error attribute accessors matching SolidQueue API
    %i[exception_class message backtrace].each do |attribute|
      define_method(attribute) { error&.with_indifferent_access&.[](attribute.to_s) }
    end

    private

    # BSON documents are limited to 16 MB. Reserve a generous budget for the
    # backtrace so a deep stack can never blow the limit.
    BACKTRACE_SIZE_LIMIT = 50_000 # bytes of JSON

    def expand_error_details_from_exception
      return unless exception

      self.error = {
        "exception_class" => exception.class.name,
        "message" => exception.message,
        "backtrace" => truncate_backtrace(exception.backtrace)
      }
    end

    def truncate_backtrace(lines)
      return lines if lines.nil?

      truncated = []
      lines.each do |line|
        truncated << line
        break if truncated.to_json.bytesize > BACKTRACE_SIZE_LIMIT
      end
      # Remove the last line that pushed us over the limit
      truncated.pop if truncated.to_json.bytesize > BACKTRACE_SIZE_LIMIT
      truncated
    end
  end
end

---

failed_execution/batchable.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class FailedExecution
    # A FailedExecution is created only after retries are exhausted, when the
    # job stops counting as pending in its batch.
    module Batchable
      extend ActiveSupport::Concern

      included do
        after_create :destroy_job_batch_execution, if: -> { Batch.migrated? && job.batch_id? }
      end

      private
        def destroy_job_batch_execution
          job.batch_execution&.destroy!
        rescue ActiveRecord::ActiveRecordError => e
          SolidQueue.instrument(:batch_progress_error, batch_id: job.batch_id, job_id: job.id, error: e)
        end
    end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/failed_execution/batchable.rb`

No local equivalent exists for this file.

---

job.rb — 🔄 Modified upstream

📥 What changed in upstream (v1.6.0 → v1.7.0)

--- solid_queue@v1.6.0/job.rb
+++ solid_queue@v1.7.0/job.rb
@@ -4,13 +4,19 @@
   class Job < Record
     class EnqueueError < StandardError; end
 
-    include Executable, Clearable, Recurrable
+    include Executable, Clearable, Recurrable, Batchable
 
     serialize :arguments, coder: JSON
 
     class << self
       def enqueue_all(active_jobs)
-        active_jobs.each { |job| job.scheduled_at ||= Time.current }
+        # Bulk enqueues bypass ActiveJob#enqueue, so batch membership is captured here
+        current_batch_id = Batch.current_batch_id
+
+        active_jobs.each do |job|
+          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)
 
         transaction do
@@ -63,7 +69,9 @@
             class_name: active_job.class.name,
             arguments: active_job.serialize,
             concurrency_key: active_job.concurrency_key
-          }
+          }.tap do |attributes|
+            attributes[:batch_id] = active_job.batch_id if Batch.migrated?
+          end
         end
     end
   end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/job.rb`

# frozen_string_literal: true

module SolidQueue
  class Job < Record
    class EnqueueError < StandardError; end

    include Clearable
    include Recurrable
    include Executable # includes ConcurrencyControls, Schedulable, Retryable

    field :queue_name, type: String
    field :class_name, type: String
    field :arguments, type: Hash, default: {}
    field :priority, type: Integer, default: 0
    field :active_job_id, type: String
    field :concurrency_key, type: String
    field :concurrency_limit, type: Integer # stored per-job; overrides job_class.concurrency_limit
    field :finished_at, type: Time
    field :max_retries, type: Integer, default: 0
    field :retry_count, type: Integer, default: 0

    index({ queue_name: 1 })
    index({ class_name: 1 })
    index({ priority: 1 })
    index({ active_job_id: 1 }, { sparse: true })
    index({ finished_at: 1 }, { sparse: true })
    index({ concurrency_key: 1 }, { sparse: true })

    validates :queue_name, :class_name, presence: true

    DEFAULT_PRIORITY = 0
    DEFAULT_QUEUE_NAME = "default"

    class << self
      # Primary enqueue entry point — called by the ActiveJob adapter.
      def enqueue(active_job, scheduled_at: Time.current)
        active_job.scheduled_at = scheduled_at

        create_from_active_job(active_job).tap do |enqueued_job|
          if enqueued_job.persisted?
            active_job.provider_job_id = enqueued_job.id.to_s
            active_job.successfully_enqueued = true
          end
        end
      end

      # Bulk enqueue — called by the ActiveJob adapter for perform_all_later.
      def enqueue_all(active_jobs)
        active_jobs.each { |job| job.scheduled_at = Time.current }
        active_jobs_by_job_id = active_jobs.index_by(&:job_id)

        jobs = create_all_from_active_jobs(active_jobs)

        prepare_all_for_execution(jobs).tap do |enqueued_jobs|
          enqueued_jobs.each do |enqueued_job|
            aj = active_jobs_by_job_id[enqueued_job.active_job_id]
            next unless aj

            aj.provider_job_id = enqueued_job.id.to_s
            aj.successfully_enqueued = true
          end
        end

        active_jobs.count(&:successfully_enqueued?)
      end

      private

      def create_from_active_job(active_job)
        create!(**attributes_from_active_job(active_job))
      rescue StandardError => e
        enqueue_error = EnqueueError.new("#{e.class.name}: #{e.message}").tap do |err|
          err.set_backtrace(e.backtrace)
        end
        raise enqueue_error
      end

      def create_all_from_active_jobs(active_jobs)
        active_jobs.filter_map do |active_job|
          create_from_active_job(active_job)
        rescue EnqueueError
          nil
        end
      end

      def attributes_from_active_job(active_job)
        {
          queue_name: active_job.queue_name || DEFAULT_QUEUE_NAME,
          active_job_id: active_job.job_id,
          priority: active_job.priority || DEFAULT_PRIORITY,
          scheduled_at: active_job.scheduled_at,
          class_name: active_job.class.name,
          arguments: active_job.serialize,
          concurrency_key: active_job.try(:concurrency_key)
        }.compact
      end
    end

    def deserialize_for_active_job
      ActiveJob::Base.deserialize(arguments)
    end
  end
end

---

job/batchable.rb — 🆕 New upstream file

📥 What changed in upstream (v1.6.0 → v1.7.0)

# frozen_string_literal: true

module SolidQueue
  class Job
    module Batchable
      extend ActiveSupport::Concern

      included do
        belongs_to :batch, optional: true
        has_one :batch_execution

        after_create :create_batch_execution, if: :batched?
        after_update :update_batch_progress, if: :batched?
        before_destroy :destroy_batch_execution, if: :batched?
      end

      class_methods do
        def batch_all(jobs)
          BatchExecution.create_all_from_jobs(jobs) if Batch.migrated?
        end
      end

      # Also guards against the batches schema not being installed: without
      # its migration, jobs don't even have a batch_id.
      def batched?
        Batch.migrated? && batch_id?
      end

      private
        def create_batch_execution
          BatchExecution.create_all_from_jobs([ self ])
        end

        def update_batch_progress
          return unless saved_change_to_finished_at? && finished_at.present?

          batch_execution&.destroy!
        rescue ActiveRecord::ActiveRecordError => e
          SolidQueue.instrument(:batch_progress_error, batch_id: batch_id, job_id: id, error: e)
        end

        # Destroy through Active Record instead of relying on the foreign
        # key's cascade, so destroying the tracking row retries the batch
        # completion check.
        def destroy_batch_execution
          batch_execution&.destroy!
        end
    end
  end
end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/job/batchable.rb`

No local equivalent exists for this file.

---

job/executable.rb — 🔄 Modified upstream

📥 What changed in upstream (v1.6.0 → v1.7.0)

--- solid_queue@v1.6.0/job/executable.rb
+++ solid_queue@v1.7.0/job/executable.rb
@@ -18,6 +18,9 @@
 
       class_methods do
         def prepare_all_for_execution(jobs)
+          # Track before dispatch so conflict-discarded jobs count like single enqueues.
+          batch_all(jobs)
+
           due, not_yet_due = jobs.partition(&:due?)
           dispatch_all(due) + schedule_all(not_yet_due)
         end
@@ -78,7 +81,8 @@
 
       def finished!
         if SolidQueue.preserve_finished_jobs?
-          touch(:finished_at)
+          # update! rather than touch so the batch tracking callbacks run
+          update!(finished_at: Time.current)
         else
           destroy!
         end

📄 Our current Mongoid model — `lib/solid_queue_mongoid/models/job/executable.rb`

# frozen_string_literal: true

module SolidQueue
  class Job
    module Executable
      extend ActiveSupport::Concern

      included do
        include ConcurrencyControls, Schedulable, Retryable

        has_one :ready_execution,   class_name: "SolidQueue::ReadyExecution",   dependent: :destroy
        has_one :claimed_execution, class_name: "SolidQueue::ClaimedExecution", dependent: :destroy

        after_create :prepare_for_execution

        scope :finished, -> { where(:finished_at.ne => nil) }
        scope :failed,   -> { where(:id.in => SolidQueue::FailedExecution.all.pluck(:job_id)) }
        scope :pending,  -> { where(finished_at: nil) }
      end

      class_methods do # rubocop:disable Metrics/BlockLength
        # Dispatch a collection of jobs, partitioned by schedule and concurrency.
        def prepare_all_for_execution(jobs)
          due, not_yet_due = jobs.partition(&:due?)
          dispatch_all(due) + schedule_all(not_yet_due)
        end

        def dispatch_all(jobs)
          with_concurrency_limits, without_concurrency_limits = jobs.partition(&:concurrency_limited?)

          dispatch_all_at_once(without_concurrency_limits)
          dispatch_all_one_by_one(with_concurrency_limits)

          successfully_dispatched(jobs)
        end

        private

        def dispatch_all_at_once(jobs)
          ReadyExecution.create_all_from_jobs(jobs)
        end

        def dispatch_all_one_by_one(jobs)
          jobs.each(&:dispatch)
        end

        def successfully_dispatched(jobs)
          jobs.map(&:id)
          dispatched_and_ready(jobs) + dispatched_and_blocked(jobs)
        end

        def dispatched_and_ready(jobs)
          job_ids = jobs.map(&:id)
          where(:id.in => ReadyExecution.where(:job_id.in => job_ids).pluck(:job_id))
        end

        def dispatched_and_blocked(jobs)
          job_ids = jobs.map(&:id)
          where(:id.in => BlockedExecution.where(:job_id.in => job_ids).pluck(:job_id))
        end
      end

      # status helpers matching SolidQueue runtime expectations
      %w[ready claimed failed].each do |status|
        define_method("#{status}?") { public_send("#{status}_execution").present? }
      end

      def prepare_for_execution
        if due?
          dispatch
        else
          schedule
        end
      end

      def dispatch
        if due?
          if acquire_concurrency_lock
            ready
          else
            handle_concurrency_conflict
          end
        else
          schedule
        end
      end

      # Called by ClaimedExecution#release — bypasses the semaphore check.
      def dispatch_bypassing_concurrency_limits
        ready
      end

      def finished!
        if SolidQueue.preserve_finished_jobs?
          update(finished_at: Time.current)
          # Clean up the claimed execution if still present (e.g. called directly
          # outside of ClaimedExecution#finished which does its own destroy!).
          claimed_execution&.destroy
        else
          destroy!
        end
      end

      alias finish finished!

      def finished?
        finished_at.present?
      end

      def status
        if finished?
          :finished
        elsif (exec = execution)
          exec.type
        end
      end

      def discard
        execution&.discard
      end

      def ready
        existing = ReadyExecution.where(job_id: id).first
        return existing if existing

        re = ReadyExecution.new(job_id: id)
        re.queue_name = queue_name
        re.priority   = priority
        re.save!
        re
      rescue Mongoid::Errors::Validations, Mongo::Error::OperationFailure
        ReadyExecution.where(job_id: id).first
      end

      def execution
        %w[ready claimed failed].reduce(nil) do |acc, status|
          acc || public_send("#{status}_execution")
        end
      end
    end
  end
end

[github-actions]

@copilot Please review the upstream diff above and update the corresponding Mongoid models in lib/solid_queue_mongoid/models/ to reflect any behavioural changes. Do not copy ActiveRecord patterns directly — adapt them to Mongoid conventions used in this gem.