Skip to content

Dynamic scheduled tasks #553

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Dynamic scheduled tasks #553

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -607,6 +607,8 @@ Rails.application.config.after_initialize do # or to_prepare
end
```

You can also dynamically add or remove recurring tasks by creating or deleting SolidQueue::RecurringTask records. It works the same way as with static tasks, except you must set the static field to false. Changes won’t be picked up immediately — they take effect after about a one-minute delay.

It's possible to run multiple schedulers with the same `recurring_tasks` configuration, for example, if you have multiple servers for redundancy, and you run the `scheduler` in more than one of them. To avoid enqueuing duplicate tasks at the same time, an entry in a new `solid_queue_recurring_executions` table is created in the same transaction as the job is enqueued. This table has a unique index on `task_key` and `run_at`, ensuring only one entry per task per time will be created. This only works if you have `preserve_finished_jobs` set to `true` (the default), and the guarantee applies as long as you keep the jobs around.

**Note**: a single recurring schedule is supported, so you can have multiple schedulers using the same schedule, but not multiple schedulers using different configurations.
Expand Down
1 change: 1 addition & 0 deletions app/models/solid_queue/recurring_task.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ class RecurringTask < Record
validate :existing_job_class

scope :static, -> { where(static: true) }
scope :dynamic, -> { where(static: false) }

has_many :recurring_executions, foreign_key: :task_key, primary_key: :key

Expand Down
10 changes: 5 additions & 5 deletions lib/solid_queue/configuration.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ def default_options
end

def invalid_tasks
recurring_tasks.select(&:invalid?)
static_recurring_tasks.select(&:invalid?)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't need to change names. It's clear the tasks here are static since this comes from the recurring.yml configuration. We don't need to rename anything here.

end

def only_work?
Expand Down Expand Up @@ -122,8 +122,8 @@ def dispatchers
end

def schedulers
if !skip_recurring_tasks? && recurring_tasks.any?
[ Process.new(:scheduler, recurring_tasks: recurring_tasks) ]
if !skip_recurring_tasks?
[ Process.new(:scheduler, recurring_tasks: static_recurring_tasks) ]
else
[]
end
Expand All @@ -139,8 +139,8 @@ def dispatchers_options
.map { |options| options.dup.symbolize_keys }
end

def recurring_tasks
@recurring_tasks ||= recurring_tasks_config.map do |id, options|
def static_recurring_tasks
@static_recurring_tasks ||= recurring_tasks_config.map do |id, options|
RecurringTask.from_configuration(id, **options) if options&.has_key?(:schedule)
end.compact
end
Expand Down
6 changes: 6 additions & 0 deletions lib/solid_queue/scheduler.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,12 @@ def run
loop do
break if shutting_down?

recurring_schedule.update_scheduled_tasks.tap do |updated_tasks|
if updated_tasks.any?
process.update_columns(metadata: metadata.compact)
end
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This code is mixing actions at very different levels, making it aware of details it shouldn't need to know, like how to update the metadata for its registered process record, or whether the recurring schedule changed. It should change, perhaps to something like

recurring_schedule.reload!
if recurring_schedule.changed?
  refresh_registered_process
end

And refresh_registered_process would go in SolidQueue::Processes::Registrable.

end

interruptible_sleep(SLEEP_INTERVAL)
end
ensure
Expand Down
46 changes: 36 additions & 10 deletions lib/solid_queue/scheduler/recurring_schedule.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@ module SolidQueue
class Scheduler::RecurringSchedule
include AppExecutor

attr_reader :configured_tasks, :scheduled_tasks
attr_reader :static_tasks, :configured_tasks, :scheduled_tasks

def initialize(tasks)
@configured_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }.select(&:valid?)
@static_tasks = Array(tasks).map { |task| SolidQueue::RecurringTask.wrap(task) }.select(&:valid?)
@configured_tasks = @static_tasks + dynamic_tasks
@scheduled_tasks = Concurrent::Hash.new
end

Expand All @@ -17,15 +18,36 @@ def empty?

def schedule_tasks
wrap_in_app_executor do
persist_tasks
reload_tasks
persist_static_tasks
reload_static_tasks
end

configured_tasks.each do |task|
schedule_task(task)
end
end

def dynamic_tasks
SolidQueue::RecurringTask.dynamic
end

def schedule_new_dynamic_tasks
dynamic_tasks.where.not(key: scheduled_tasks.keys).each do |task|
schedule_task(task)
end
end

def unschedule_old_dynamic_tasks
(scheduled_tasks.keys - SolidQueue::RecurringTask.pluck(:key)).each do |key|
scheduled_tasks[key].cancel
scheduled_tasks.delete(key)
end
end

def update_scheduled_tasks
schedule_new_dynamic_tasks + unschedule_old_dynamic_tasks
end

def schedule_task(task)
scheduled_tasks[task.key] = schedule(task)
end
Expand All @@ -35,18 +57,22 @@ def unschedule_tasks
scheduled_tasks.clear
end

def static_task_keys
static_tasks.map(&:key)
end

def task_keys
configured_tasks.map(&:key)
static_task_keys + dynamic_tasks.map(&:key)
end

private
def persist_tasks
SolidQueue::RecurringTask.static.where.not(key: task_keys).delete_all
SolidQueue::RecurringTask.create_or_update_all configured_tasks
def persist_static_tasks
SolidQueue::RecurringTask.static.where.not(key: static_task_keys).delete_all
SolidQueue::RecurringTask.create_or_update_all static_tasks
end

def reload_tasks
@configured_tasks = SolidQueue::RecurringTask.where(key: task_keys)
def reload_static_tasks
@static_tasks = SolidQueue::RecurringTask.static.where(key: static_task_keys)
end

def schedule(task)
Expand Down
6 changes: 3 additions & 3 deletions test/unit/configuration_test.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ class ConfigurationTest < ActiveSupport::TestCase
test "default configuration when config given is empty" do
configuration = SolidQueue::Configuration.new(config_file: config_file_path(:empty_configuration), recurring_schedule_file: config_file_path(:empty_configuration))

assert_equal 2, configuration.configured_processes.count
assert_equal 3, configuration.configured_processes.count # includes scheduler for dynamic tasks
assert_processes configuration, :worker, 1, queues: "*"
assert_processes configuration, :dispatcher, 1, batch_size: SolidQueue::Configuration::DISPATCHER_DEFAULTS[:batch_size]
end
Expand Down Expand Up @@ -101,11 +101,11 @@ class ConfigurationTest < ActiveSupport::TestCase

configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_production_only))
assert configuration.valid?
assert_processes configuration, :scheduler, 0
assert_processes configuration, :scheduler, 1 # Starts in case of dynamic tasks

configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_empty))
assert configuration.valid?
assert_processes configuration, :scheduler, 0
assert_processes configuration, :scheduler, 1 # Starts in case of dynamic tasks

# No processes
configuration = SolidQueue::Configuration.new(skip_recurring: true, dispatchers: [], workers: [])
Expand Down
100 changes: 99 additions & 1 deletion test/unit/scheduler_test.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
class SchedulerTest < ActiveSupport::TestCase
self.use_transactional_tests = false

test "recurring schedule" do
test "recurring schedule (only static)" do
recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }
scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks).tap(&:start)

Expand All @@ -17,6 +17,41 @@ class SchedulerTest < ActiveSupport::TestCase
scheduler.stop
end

test "recurring schedule (only dynamic)" do
SolidQueue::RecurringTask.create(
key: "dynamic_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
)
scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}).tap(&:start)

wait_for_registered_processes(1, timeout: 1.second)

process = SolidQueue::Process.first
assert_equal "Scheduler", process.kind

assert_metadata process, recurring_schedule: [ "dynamic_task" ]
ensure
scheduler.stop
end

test "recurring schedule (static + dynamic)" do
SolidQueue::RecurringTask.create(
key: "dynamic_task", static: false, class_name: "AddToBufferJob", schedule: "every second", arguments: [ 42 ]
)

recurring_tasks = { static_task: { class: "AddToBufferJob", schedule: "every hour", args: 42 } }

scheduler = SolidQueue::Scheduler.new(recurring_tasks: recurring_tasks).tap(&:start)

wait_for_registered_processes(1, timeout: 1.second)

process = SolidQueue::Process.first
assert_equal "Scheduler", process.kind

assert_metadata process, recurring_schedule: [ "static_task", "dynamic_task" ]
ensure
scheduler.stop
end

test "run more than one instance of the scheduler with recurring tasks" do
recurring_tasks = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
schedulers = 2.times.collect do
Expand All @@ -33,4 +68,67 @@ class SchedulerTest < ActiveSupport::TestCase
assert_equal 1, run_at_times[i + 1] - run_at_times[i]
end
end

test "updates metadata after adding dynamic task post-start" do
scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}).tap do |s|
s.define_singleton_method(:interruptible_sleep) { |interval| sleep 0.1 }
s.start
end

wait_for_registered_processes(1, timeout: 1.second)

process = SolidQueue::Process.first
# initially there are no recurring_schedule keys
assert process.metadata, {}

# now create a dynamic task after the scheduler has booted
SolidQueue::RecurringTask.create(
key: "new_dynamic_task",
static: false,
class_name: "AddToBufferJob",
schedule: "every second",
arguments: [ 42 ]
)

sleep 1

process.reload

# metadata should now include the new key
assert_metadata process, recurring_schedule: [ "new_dynamic_task" ]
ensure
scheduler&.stop
end

test "updates metadata after removing dynamic task post-start" do
old_dynamic_task = SolidQueue::RecurringTask.create(
key: "old_dynamic_task",
static: false,
class_name: "AddToBufferJob",
schedule: "every second",
arguments: [ 42 ]
)

scheduler = SolidQueue::Scheduler.new(recurring_tasks: {}).tap do |s|
s.define_singleton_method(:interruptible_sleep) { |interval| sleep 0.1 }
s.start
end

wait_for_registered_processes(1, timeout: 1.second)

process = SolidQueue::Process.first
# initially there is one recurring_schedule key
assert_metadata process, recurring_schedule: [ "old_dynamic_task" ]

old_dynamic_task.destroy

sleep 1

process.reload

# The task is unschedule after it's being removed, and it's reflected in the metadata
assert process.metadata, {}
ensure
scheduler&.stop
end
end