Support passing kwargs as last element of the arguments array

If we don't explicitly add a ruby2_keywords flag, Active Job will any hash
included in the arguments array with keys as `_aj_symbol_keys`, and when
deserialized, it'd be treated always as a hash argument instead of keyword
arguments. Depending on the job, this might work fine, but if the job uses
keyword arguments, trying to execute the job with deserialized arguments
will fail. However, the opposite is not true: if the job accepts a hash
argument and we pass a hash with the ruby2_keywords flag, it'll work just
fine as Active Job will serialize that with keys as `_aj_ruby2_keywords`,
so we take advantage of that to simplify the task definition and not
having to distinguish between args and kwargs.

Author: rosa

lib/solid_queue/dispatcher/recurring_task.rb

@@ -42,7 +42,11 @@ def valid?
     end
 
     def to_s
-      "#{class_name}.perform_later(#{arguments.map(&:inspect).join(",")}) [ #{schedule.original} ]"
+      "#{class_name}.perform_later(#{arguments.map(&:inspect).join(",")}) [ #{parsed_schedule} ]"
+    end
+
+    def parsed_schedule
+      schedule.original.to_s
     end
 
     private
@@ -55,7 +59,15 @@ def perform_later_and_record(run_at:)
       end
 
       def perform_later
-        job_class.perform_later(*arguments)
+        job_class.perform_later(*arguments_with_kwargs)
+      end
+
+      def arguments_with_kwargs
+        if arguments.last.is_a?(Hash)
+          arguments[0...-1] + [ Hash.ruby2_keywords_hash(arguments.last) ]
+        else
+          arguments
+        end
       end
 
       def job_class

test/dummy/app/jobs/store_result_job.rb

@@ -1,13 +1,13 @@
 class StoreResultJob < ApplicationJob
   queue_as :background
 
-  def perform(value, pause: nil, exception: nil, exit: nil)
+  def perform(value, status: :completed, pause: nil, exception: nil, exit: nil)
     result = JobResult.create!(queue_name: queue_name, status: "started", value: value)
 
     sleep(pause) if pause
     raise exception.new if exception
     exit! if exit
 
-    result.update!(status: "completed")
+    result.update!(status: status)
   end
 end

test/dummy/config/solid_queue.yml

@@ -8,9 +8,9 @@ default: &default
     - polling_interval: 1
       batch_size: 500
       recurring_tasks:
-        periodic-add-to-buffer:
-          class: AddToBufferJob
-          args: 42
+        periodic_store_result:
+          class: StoreResultJob
+          args: [ 42, { status: "custom_status" } ]
           schedule: every second
 
 development:

test/integration/recurring_tasks_test.rb

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+require "test_helper"
+
+class RecurringTasksTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  setup do
+    @pid = run_supervisor_as_fork(mode: :all)
+    # 1 supervisor + 2 workers + 1 dispatcher
+    wait_for_registered_processes(4, timeout: 3.second)
+  end
+
+  teardown do
+    terminate_process(@pid) if process_exists?(@pid)
+
+    SolidQueue::Process.destroy_all
+    SolidQueue::Job.destroy_all
+    JobResult.delete_all
+  end
+
+  test "enqueue and process periodic tasks" do
+    wait_for_jobs_to_be_enqueued(2, timeout: 2.seconds)
+    wait_for_jobs_to_finish_for(2.seconds)
+
+    terminate_process(@pid)
+
+    skip_active_record_query_cache do
+      assert SolidQueue::Job.count >= 2
+      SolidQueue::Job.all.each do |job|
+        assert_equal "periodic_store_result", job.recurring_execution.task_key
+        assert_equal "StoreResultJob", job.class_name
+      end
+
+      assert_equal 2, JobResult.count
+      JobResult.all.each do |result|
+        assert_equal "custom_status", result.status
+        assert_equal "42", result.value
+      end
+    end
+  end
+
+  private
+    def wait_for_jobs_to_be_enqueued(count, timeout: 1.second)
+      wait_while_with_timeout(timeout) { SolidQueue::Job.count < count }
+    end
+end

test/unit/dispatcher_test.rb

@@ -12,6 +12,8 @@ class DispatcherTest < ActiveSupport::TestCase
 
   teardown do
     @dispatcher.stop
+    SolidQueue::Job.delete_all
+    SolidQueue::Process.delete_all
   end
 
   test "dispatcher is registered as process" do
@@ -79,7 +81,7 @@ class DispatcherTest < ActiveSupport::TestCase
     SolidQueue.silence_polling = old_silence_polling
   end
 
-  test "run more than one instance of the dispatcher" do
+  test "run more than one instance of the dispatcher without recurring tasks" do
     15.times do
       AddToBufferJob.set(wait: 0.2).perform_later("I'm scheduled")
     end
@@ -97,4 +99,23 @@ class DispatcherTest < ActiveSupport::TestCase
 
     another_dispatcher.stop
   end
+
+  test "run more than one instance of the dispatcher with recurring tasks" do
+    recurring_task = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
+    dispatchers = 2.times.collect do
+      SolidQueue::Dispatcher.new(concurrency_maintenance: false, recurring_tasks: recurring_task)
+    end
+
+    dispatchers.each(&:start)
+    sleep 2
+    dispatchers.each(&:stop)
+
+    assert_equal SolidQueue::Job.count, SolidQueue::RecurringExecution.count
+    assert SolidQueue::Job.count < 4
+
+    run_at_times = SolidQueue::RecurringExecution.all.map(&:run_at).sort
+    0.upto(run_at_times.length - 2) do |i|
+      assert_equal 1, run_at_times[i + 1] - run_at_times[i]
+    end
+  end
 end

test/unit/recurring_task_test.rb

@@ -0,0 +1,96 @@
+require "test_helper"
+
+class RecurringTaskTest < ActiveSupport::TestCase
+  class JobWithoutArguments < ApplicationJob
+    def perform
+      JobBuffer.add "job_without_arguments"
+    end
+  end
+
+  class JobWithRegularHashArguments < ApplicationJob
+    def perform(value, options)
+      JobBuffer.add [ value, options[:value] ]
+    end
+  end
+
+  class JobWithKeywordArgument < ApplicationJob
+    def perform(value, value_kwarg:)
+      JobBuffer.add [ value, value_kwarg ]
+    end
+  end
+
+  class JobWithMultipleTypeArguments < ApplicationJob
+    def perform(value, options = {}, **kwargs)
+      JobBuffer.add [ value, options[:value], kwargs[:value_kwarg] ]
+    end
+  end
+
+  setup do
+    @worker = SolidQueue::Worker.new(queues: "*")
+    @worker.mode = :inline
+  end
+
+  test "job without arguments" do
+    task = recurring_task_with(class_name: "JobWithoutArguments")
+    enqueue_and_assert_performed_with_result task, "job_without_arguments"
+  end
+
+  test "job with regular hash argument" do
+    task = recurring_task_with(class_name: "JobWithRegularHashArguments", args: [ "regular_hash_argument", { value: 42, not_used: 24 } ])
+
+    enqueue_and_assert_performed_with_result task, [ "regular_hash_argument", 42 ]
+  end
+
+  test "job with keyword argument" do
+    task = recurring_task_with(class_name: "JobWithKeywordArgument", args: [ "keyword_argument", { value_kwarg: [ 42, 24 ] } ])
+    enqueue_and_assert_performed_with_result task, [ "keyword_argument", [ 42, 24 ] ]
+  end
+
+  test "job with arguments of multiple types" do
+    task = recurring_task_with(class_name: "JobWithMultipleTypeArguments", args:
+      [ "multiple_types", { value: "regular_hash_value", not_used: 28 }, value_kwarg: 42, not_used: 32 ])
+    enqueue_and_assert_performed_with_result task, [ "multiple_types", "regular_hash_value", 42 ]
+  end
+
+  test "job with arguments of multiple types ignoring optional regular hash" do
+    task = recurring_task_with(class_name: "JobWithMultipleTypeArguments", args:
+      [ "multiple_types", value: "regular_hash_value", value_kwarg: 42, not_used: 32 ])
+    enqueue_and_assert_performed_with_result task, [ "multiple_types", nil, 42 ]
+  end
+
+  test "valid and invalid schedules" do
+    assert_not recurring_task_with(class_name: "JobWithoutArguments", schedule: "once a year").valid?
+    assert_not recurring_task_with(class_name: "JobWithoutArguments", schedule: "tomorrow").valid?
+
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every Thursday at 1 AM")
+    assert task.valid?
+    # At 1 AM on the 4th day of the week
+    assert_equal "0 1 * * 4", task.parsed_schedule
+
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every month")
+    assert task.valid?
+    # At 12:00 AM, on day 1 of the month
+    assert_equal "0 0 1 * *", task.parsed_schedule
+
+    task = recurring_task_with(class_name: "JobWithoutArguments", schedule: "every second")
+    assert task.valid?
+    assert_equal "* * * * * *", task.parsed_schedule
+  end
+
+  private
+    def enqueue_and_assert_performed_with_result(task, result)
+      assert_difference [ -> { SolidQueue::Job.count }, -> { SolidQueue::ReadyExecution.count } ], +1 do
+        task.enqueue(at: Time.now)
+      end
+
+      assert_difference -> { JobBuffer.size }, +1 do
+        @worker.start
+      end
+
+      assert_equal result, JobBuffer.last_value
+    end
+
+    def recurring_task_with(class_name:, schedule: "every hour", args: nil)
+      SolidQueue::Dispatcher::RecurringTask.from_configuration("task-id", class: "RecurringTaskTest::#{class_name}", schedule: schedule, args: args)
+    end
+end