Merge branch 'main' into feature/clear-ar-connections-flag

Author: rafael-pissardo

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_queue (1.3.2)
+    solid_queue (1.4.0)
       activejob (>= 7.1)
       activerecord (>= 7.1)
       concurrent-ruby (>= 1.3.1)

README.md

@@ -17,6 +17,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Workers, dispatchers, and scheduler](#workers-dispatchers-and-scheduler)
   - [Fork vs. async mode](#fork-vs-async-mode)
 - [Configuration](#configuration)
+  - [Optional scheduler configuration](#optional-scheduler-configuration)
   - [Queue order and priorities](#queue-order-and-priorities)
   - [Queues specification and performance](#queues-specification-and-performance)
   - [Threads, processes, and signals](#threads-processes-and-signals)
@@ -31,6 +32,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Puma plugin](#puma-plugin)
 - [Jobs and transactional integrity](#jobs-and-transactional-integrity)
 - [Recurring tasks](#recurring-tasks)
+  - [Scheduling and unscheduling recurring tasks dynamically](#scheduling-and-unscheduling-recurring-tasks-dynamically)
 - [Inspiration](#inspiration)
 - [License](#license)
 
@@ -209,7 +211,7 @@ By default, Solid Queue will try to find your configuration under `config/queue.
 bin/jobs -c config/calendar.yml
 ```
 
-You can also skip all recurring tasks by setting the environment variable `SOLID_QUEUE_SKIP_RECURRING=true`. This is useful for environments like staging, review apps, or development where you don't want any recurring jobs to run. This is equivalent to using the `--skip-recurring` option with `bin/jobs`.
+You can also skip the scheduler process by setting the environment variable `SOLID_QUEUE_SKIP_RECURRING=true`. This is useful for environments like staging, review apps, or development where you don't want any recurring jobs to run. This is equivalent to using the `--skip-recurring` option with `bin/jobs`.
 
 This is what this configuration looks like:
 
@@ -227,6 +229,10 @@ production:
       threads: 5
       polling_interval: 0.1
       processes: 3
+  scheduler:
+    dynamic_tasks_enabled: true
+    polling_interval: 5
+
 ```
 
 Everything is optional. If no configuration at all is provided, Solid Queue will run with one dispatcher and one worker with default settings. If you want to run only dispatchers or workers, you just need to include that section alone in the configuration. For example, with the following configuration:
@@ -271,6 +277,19 @@ It is recommended to set this value less than or equal to the queue database's c
 - `concurrency_maintenance`: whether the dispatcher will perform the concurrency maintenance work. This is `true` by default, and it's useful if you don't use any [concurrency controls](#concurrency-controls) and want to disable it or if you run multiple dispatchers and want some of them to just dispatch jobs without doing anything else.
 
 
+### Optional scheduler configuration
+
+Optionally, you can configure the scheduler process under the `scheduler` section in your `config/queue.yml` if you'd like to [schedule recurring tasks dynamically](#scheduling-and-unscheduling-recurring-tasks-dynamically).
+
+```yaml
+scheduler:
+  dynamic_tasks_enabled: true
+  polling_interval: 5
+```
+
+- `dynamic_tasks_enabled`: whether the scheduler should poll for [dynamically scheduled recurring tasks](#scheduling-and-unscheduling-recurring-tasks-dynamically). This is `false` by default. When enabled, the scheduler will poll the database at the given `polling_interval` to pick up tasks scheduled via `SolidQueue.schedule_recurring_task`.
+- `polling_interval`: how frequently (in seconds) the scheduler checks for dynamic task changes. Defaults to `5`.
+
 ### Queue order and priorities
 
 As mentioned above, if you specify a list of queues for a worker, these will be polled in the order given, such as for the list `real_time,background`, no jobs will be taken from `background` unless there aren't any more jobs waiting in `real_time`.
@@ -462,7 +481,7 @@ class MyJob < ApplicationJob
 - `group` is used to control the concurrency of different job classes together. It defaults to the job class name.
 - `on_conflict` controls behaviour when enqueuing a job that conflicts with the concurrency limits configured. It can be set to one of the following:
   - (default) `:block`: the job is blocked and is dispatched when another job completes and unblocks it, or when the duration expires.
-  - `:discard`: the job is discarded. When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded up to the interval defined by `duration` has elapsed.
+  - `:discard`: the job is discarded. When you choose this option, bear in mind that if a job runs and fails to remove the concurrency lock (or _semaphore_, read below to know more about this), all jobs conflicting with it will be discarded until the interval defined by `duration` has elapsed.
 
 When a job includes these controls, we'll ensure that, at most, the number of jobs (indicated as `to`) that yield the same `key` will be performed concurrently, and this guarantee will last for `duration` for each job enqueued. Note that there's no guarantee about _the order of execution_, only about jobs being performed at the same time (overlapping).
 
@@ -472,7 +491,7 @@ Since something can happen that prevents the first job from releasing the semaph
 
 It's important to note that after one or more candidate jobs are unblocked (either because a job finishes or because `duration` expires and a semaphore is released), the `duration` timer for the still blocked jobs is reset. This happens indirectly via the expiration time of the semaphore, which is updated.
 
-When using `discard` as the behaviour to handle conflicts, you might have jobs discarded for up to the `duration` interval if something happens and a running job fails to release the semaphore.
+When using `discard` as the behaviour to handle conflicts, you might have jobs discarded for until the `duration` interval if something happens and a running job fails to release the semaphore.
 
 
 For example:
@@ -732,6 +751,38 @@ my_periodic_resque_job:
 
 and the job will be enqueued via `perform_later` so it'll run in Resque. However, in this case we won't track any `solid_queue_recurring_execution` record for it and there won't be any guarantees that the job is enqueued only once each time.
 
+### Scheduling and unscheduling recurring tasks dynamically
+
+You can schedule and unschedule recurring tasks at runtime, without editing the configuration file. To enable this, you need to set `dynamic_tasks_enabled: true` in the `scheduler` section of your `config/queue.yml`, [as explained earlier](#optional-scheduler-configuration).
+
+```yaml
+scheduler:
+  dynamic_tasks_enabled: true
+```
+
+Then you can use the following methods to add recurring tasks dynamically:
+
+```ruby
+SolidQueue.schedule_recurring_task(
+  "my_dynamic_task",
+  class: "MyJob",
+  args: [1, 2],
+  schedule: "every 10 minutes"
+)
+```
+
+This accepts the same options as the YAML configuration: `class`, `args`, `command`, `schedule`, `queue`, `priority`, and `description`.
+
+To remove a dynamically scheduled task:
+
+```ruby
+SolidQueue.unschedule_recurring_task("my_dynamic_task")
+```
+
+Only dynamic tasks can be unscheduled at runtime. Attempting to unschedule a static task (defined in `config/recurring.yml`) will raise an `ActiveRecord::RecordNotFound` error.
+
+Tasks scheduled like this persist between Solid Queue's restarts and won't stop running until you manually unschedule them. 
+
 ## Inspiration
 
 Solid Queue has been inspired by [resque](https://github.com/resque/resque) and [GoodJob](https://github.com/bensheldon/good_job). We recommend checking out these projects as they're great examples from which we've learnt a lot.

app/models/solid_queue/ready_execution.rb

@@ -30,7 +30,8 @@ def select_and_lock(queue_relation, process_id, limit)
         end
 
         def select_candidates(queue_relation, limit)
-          queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id)
+          # Force query execution here with #to_a to avoid unintended FOR UPDATE query executions
+          queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id).to_a
         end
 
         def lock_candidates(executions, process_id)

app/models/solid_queue/recurring_task.rb

@@ -11,6 +11,7 @@ class RecurringTask < Record
     validate :ensure_existing_job_class
 
     scope :static, -> { where(static: true) }
+    scope :dynamic, -> { where(static: false) }
 
     has_many :recurring_executions, foreign_key: :task_key, primary_key: :key
 
@@ -32,7 +33,15 @@ def from_configuration(key, **options)
           queue_name: options[:queue].presence,
           priority: options[:priority].presence,
           description: options[:description],
-          static: true
+          static: options.fetch(:static, true)
+      end
+
+      def create_dynamic_task(key, **options)
+        from_configuration(key, **options.merge(static: false)).save!
+      end
+
+      def delete_dynamic_task(key)
+        RecurringTask.dynamic.find_by!(key: key).destroy
       end
 
       def create_or_update_all(tasks)

lib/solid_queue.rb

@@ -44,6 +44,14 @@ module SolidQueue
 
   delegate :on_start, :on_stop, :on_exit, to: Supervisor
 
+  def schedule_recurring_task(key, **options)
+    RecurringTask.create_dynamic_task(key, **options)
+  end
+
+  def unschedule_recurring_task(key)
+    RecurringTask.delete_dynamic_task(key)
+  end
+
   [ Dispatcher, Scheduler, Worker ].each do |process|
     define_singleton_method(:"on_#{process.name.demodulize.downcase}_start") do |&block|
       process.on_start(&block)

lib/solid_queue/configuration.rb

@@ -28,6 +28,11 @@ def instantiate
       concurrency_maintenance_interval: 600
     }
 
+    SCHEDULER_DEFAULTS = {
+      polling_interval: 5,
+      dynamic_tasks_enabled: false
+    }
+
     DEFAULT_CONFIG_FILE_PATH = "config/queue.yml"
     DEFAULT_RECURRING_SCHEDULE_FILE_PATH = "config/recurring.yml"
 
@@ -137,8 +142,10 @@ def dispatchers
       end
 
       def schedulers
-        if !skip_recurring_tasks? && recurring_tasks.any?
-          [ Process.new(:scheduler, recurring_tasks: recurring_tasks) ]
+        return [] if skip_recurring_tasks?
+
+        if recurring_tasks.any? || dynamic_recurring_tasks_enabled?
+          [ Process.new(:scheduler, { recurring_tasks: recurring_tasks, **scheduler_options.with_defaults(SCHEDULER_DEFAULTS) }) ]
         else
           []
         end
@@ -154,17 +161,29 @@ def dispatchers_options
           .map { |options| options.dup.symbolize_keys }
       end
 
+      def scheduler_options
+        @scheduler_options ||= processes_config.fetch(:scheduler, {}).dup.symbolize_keys
+      end
+
+      def dynamic_recurring_tasks_enabled?
+        scheduler_options.fetch(:dynamic_tasks_enabled, SCHEDULER_DEFAULTS[:dynamic_tasks_enabled])
+      end
+
       def recurring_tasks
         @recurring_tasks ||= recurring_tasks_config.map do |id, options|
-          RecurringTask.from_configuration(id, **options) if options&.has_key?(:schedule)
+          RecurringTask.from_configuration(id, **options.merge(static: true)) if options&.has_key?(:schedule)
         end.compact
       end
 
       def processes_config
         @processes_config ||= config_from \
-          options.slice(:workers, :dispatchers).presence || options[:config_file],
-          keys: [ :workers, :dispatchers ],
-          fallback: { workers: [ WORKER_DEFAULTS ], dispatchers: [ DISPATCHER_DEFAULTS ] }
+          options.slice(:workers, :dispatchers, :scheduler).presence || options[:config_file],
+          keys: [ :workers, :dispatchers, :scheduler ],
+          fallback: {
+            workers: [ WORKER_DEFAULTS ],
+            dispatchers: [ DISPATCHER_DEFAULTS ],
+            scheduler: SCHEDULER_DEFAULTS
+          }
       end
 
       def recurring_tasks_config
@@ -173,7 +192,6 @@ def recurring_tasks_config
         end
       end
 
-
       def config_from(file_or_hash, keys: [], fallback: {}, env: Rails.env)
         load_config_from(file_or_hash).then do |config|
           config = config[env.to_sym] ? config[env.to_sym] : config

lib/solid_queue/processes/registrable.rb

@@ -59,5 +59,9 @@ def heartbeat
         self.process = nil
         wake_up
       end
+
+      def reload_metadata
+        wrap_in_app_executor { process&.update(metadata: metadata.compact) }
+      end
   end
 end

lib/solid_queue/scheduler.rb

@@ -5,7 +5,7 @@ class Scheduler < Processes::Base
     include Processes::Runnable
     include LifecycleHooks
 
-    attr_reader :recurring_schedule
+    attr_reader :recurring_schedule, :polling_interval
 
     after_boot :run_start_hooks
     after_boot :schedule_recurring_tasks
@@ -14,7 +14,10 @@ class Scheduler < Processes::Base
     after_shutdown :run_exit_hooks
 
     def initialize(recurring_tasks:, **options)
-      @recurring_schedule = RecurringSchedule.new(recurring_tasks)
+      options = options.dup.with_defaults(SolidQueue::Configuration::SCHEDULER_DEFAULTS)
+      @dynamic_tasks_enabled = options[:dynamic_tasks_enabled]
+      @polling_interval = options[:polling_interval]
+      @recurring_schedule = RecurringSchedule.new(recurring_tasks, dynamic_tasks_enabled: @dynamic_tasks_enabled)
 
       super(**options)
     end
@@ -24,13 +27,16 @@ def metadata
     end
 
     private
-      SLEEP_INTERVAL = 60 # Right now it doesn't matter, can be set to 1 in the future for dynamic tasks
+
+      STATIC_SLEEP_INTERVAL = 60
 
       def run
         loop do
           break if shutting_down?
 
-          interruptible_sleep(SLEEP_INTERVAL)
+          reload_dynamic_schedule if dynamic_tasks_enabled?
+
+          interruptible_sleep(sleep_interval)
         end
       ensure
         SolidQueue.instrument(:shutdown_process, process: self) do
@@ -46,10 +52,23 @@ def unschedule_recurring_tasks
         recurring_schedule.unschedule_tasks
       end
 
+      def reload_dynamic_schedule
+        recurring_schedule.reschedule_dynamic_tasks
+        reload_metadata
+      end
+
+      def dynamic_tasks_enabled?
+        @dynamic_tasks_enabled
+      end
+
       def all_work_completed?
         recurring_schedule.empty?
       end
 
+      def sleep_interval
+        dynamic_tasks_enabled? ? polling_interval : STATIC_SLEEP_INTERVAL
+      end
+
       def set_procline
         procline "scheduling #{recurring_schedule.task_keys.join(",")}"
       end