Merge pull request #592 from Shopify/shutdown-enqueue

Gracefully shut down rufus scheduler threads

Author: hkdsun

.rubocop_todo.yml

@@ -39,7 +39,7 @@ Metrics/MethodLength:
 # Offense count: 2
 # Configuration parameters: CountComments.
 Metrics/ModuleLength:
-  Max: 317
+  Max: 322
 
 # Offense count: 1
 Style/CaseEquality:

appveyor.yml

@@ -5,6 +5,7 @@ environment:
   RUBYOPT: '-W0'
   COVERAGE: 1
   JRUBY_OPTS: ''
+  TZ: 'UTC'
 
 install:
 - set PATH=C:\Ruby23\bin;%PATH%

lib/resque/scheduler.rb

@@ -367,7 +367,13 @@ def handle_signals_with_operation
         true
       end
 
+      def stop_rufus_scheduler
+        rufus_scheduler.shutdown(:wait)
+        rufus_scheduler.join
+      end
+
       def before_shutdown
+        stop_rufus_scheduler
         release_master_lock
       end
 

resque-scheduler.gemspec

@@ -47,6 +47,7 @@ Gem::Specification.new do |spec|
   spec.add_development_dependency 'simplecov'
   spec.add_development_dependency 'test-unit'
   spec.add_development_dependency 'yard'
+  spec.add_development_dependency 'tzinfo-data'
 
   # We pin rubocop because new cops have a tendency to result in false-y
   # positives for new contributors, which is not a nice experience.

test/scheduler_test.rb

@@ -12,6 +12,7 @@
     Resque.redis.flushall
     Resque::Scheduler.clear_schedule!
     Resque::Scheduler.send(:instance_variable_set, :@scheduled_jobs, {})
+    Resque::Scheduler.send(:instance_variable_set, :@shutdown, false)
   end
 
   test 'enqueue constantizes' do
@@ -456,6 +457,63 @@
            /#{Resque::Scheduler.send(:internal_name)}: cage/
   end
 
+  test 'gracefully shuts down rufus-scheduler threads' do
+    if RUBY_ENGINE == 'jruby' || RUBY_PLATFORM =~ /mingw|windows/i
+      omit("forking is not supported on #{RUBY_ENGINE}/#{RUBY_PLATFORM} but " \
+           'this behaviour is best tested using forks')
+    end
+
+    class BeforeEnqueueJob
+      @queue = :quick
+
+      class << self
+        def before_enqueue_example(*)
+          return false if enqueue_started?
+          enqueue_started!
+
+          sleep 5
+          true
+        end
+
+        def enqueue_started?
+          Resque.redis.get('before_enqueue_job:enqueued') == 'true'
+        end
+
+        def perform(*)
+        end
+
+        private
+
+        def enqueue_started!
+          Resque.redis.set('before_enqueue_job:enqueued', 'true')
+        end
+      end
+    end
+
+    schedule = {
+      'BeforeEnqueueJob' => { cron: '* * * * * *', class: 'BeforeEnqueueJob' }
+    }
+
+    pid = fork do
+      Resque::Scheduler.clear_schedule!
+      Resque.schedule = schedule
+      Resque::Scheduler.run
+    end
+
+    begin
+      30.times do
+        break if BeforeEnqueueJob.enqueue_started?
+        sleep 0.1
+      end
+    ensure
+      Process.kill('TERM', pid)
+      Process.wait(pid)
+    end
+
+    assert BeforeEnqueueJob.enqueue_started?, "Job enqueue didn't start in time"
+    assert_equal 1, Resque.size('quick')
+  end
+
   context 'printing schedule' do
     setup do
       Resque::Scheduler.expects(:log!).at_least_once