Merge 84882df into d74b3ce

lib/delayed/heartbeat/worker.rb

@@ -38,6 +38,11 @@ def unlock_jobs(mark_attempt_failed: true)
         orphaned_jobs
       end
 
+      # Unlocks jobs held by the worker as an atomic operation.
+      def clear_locks
+        jobs.update_all(locked_at: nil, locked_by: nil)
+      end
+
       def self.dead_workers(timeout_seconds)
         where('last_heartbeat_at < ?', Time.now.utc - timeout_seconds)
       end

lib/delayed/heartbeat/worker_heartbeat.rb

@@ -54,9 +54,20 @@ def run_heartbeat_loop
         exit(false)
       ensure
         @stop_reader.close
-        @worker_model.delete
-        # Note: The built-in Delayed::Plugins::ClearLocks will unlock the jobs for us
-        Delayed::Backend::ActiveRecord::Job.clear_active_connections!
+
+        # Note: The built-in Delayed::Plugins::ClearLocks will try to
+        # unlock the jobs for us, but in some cases this may fail
+        # (e.g. the worker is interrupted in a critical section and
+        # the database connection is in an unusable state). Here we
+        # first close all existing connections to release any locks on
+        # the delayed jobs/workers table, then reconnect with a known
+        # good connection to clear locks on jobs held by this worker.
+        Delayed::Backend::ActiveRecord::Job.clear_all_connections!
+
+        Delayed::Backend::ActiveRecord::Job.transaction do
+          @worker_model.clear_locks
+          @worker_model.delete
+        end
       end
 
       def update_heartbeat

spec/db/database.yml

@@ -2,4 +2,9 @@ sqlite3:
   adapter: sqlite3
   pool: 5
   timeout: 15000
+  # The checkout_timeout determines how long we wait before forcing a
+  # disconnect on an in-use connection. We rely on this behavior when
+  # testing shutting down workers, so this is set to a small value to
+  # ensure the tests run fast.
+  checkout_timeout: 0.01
   database: tmp/sqlite3.db

spec/delayed/heartbeat/worker_heartbeat_spec.rb

@@ -39,6 +39,11 @@
     it "destroys the worker model" do
       expect(find_worker_model(worker_name)).not_to be_present
     end
+
+    it "unlocks the worker jobs" do
+      expect(job.reload.locked_by).to be_nil
+      expect(job.reload.locked_at).to be_nil
+    end
   end
 
   context "when the heartbeat times out" do