Raise within any workers which haven't finished within the hard timeout, fixes #377

lib/sidekiq/cli.rb

@@ -9,6 +9,11 @@ require 'sidekiq'
 require 'sidekiq/util'
 
 module Sidekiq
+  # Used to raise in workers that have not finished within the
+  # hard timeout limit.  This is needed to rollback db transactions,
+  # otherwise Ruby's Thread#kill will commit.  See #377.
+  class Shutdown < RuntimeError; end
+
   class CLI
     include Util
     include Singleton

lib/sidekiq/manager.rb

@@ -138,7 +138,8 @@ module Sidekiq
 
           logger.debug { "Terminating worker threads" }
           @busy.each do |processor|
-            processor.terminate if processor.alive?
+            t = processor.bare_object.actual_work_thread
+            t.raise Shutdown if processor.alive?
           end
 
           after(0) { signal(:shutdown) }

lib/sidekiq/middleware/server/retry_jobs.rb

@@ -48,6 +48,8 @@ module Sidekiq
 
         def call(worker, msg, queue)
           yield
+        rescue Sidekiq::Shutdown
+          raise
         rescue Exception => e
           raise e unless msg['retry']
           max_retry_attempts = retry_attempts_from(msg['retry'], DEFAULT_MAX_RETRY_ATTEMPTS)

lib/sidekiq/processor.rb

@@ -26,6 +26,11 @@ module Sidekiq
       end
     end
 
+    # store the actual working thread so we
+    # can later kill if it necessary during
+    # hard shutdown.
+    attr_accessor :actual_work_thread
+
     def initialize(boss)
       @boss = boss
     end
@@ -34,6 +39,7 @@ module Sidekiq
       msgstr = work.message
       queue = work.queue_name
       defer do
+        @actual_work_thread = Thread.current
         begin
           msg = Sidekiq.load_json(msgstr)
           klass  = msg['class'].constantize
@@ -45,6 +51,9 @@ module Sidekiq
               worker.perform(*cloned(msg['args']))
             end
           end
+        rescue Sidekiq::Shutdown
+          # Had to force kill this job because it didn't finish
+          # within the timeout.
         rescue Exception => ex
           handle_exception(ex, msg || { :message => msgstr })
           raise