Fix smartproxy worker heartbeat thread

app/models/miq_queue_worker_base/runner.rb

@@ -78,7 +78,6 @@ def get_message_via_sql
         :priority   => @worker.class.queue_priority
       )
       return msg unless msg == :stale
-      heartbeat
     end
   end
 

app/models/miq_smart_proxy_worker/runner.rb

@@ -1,8 +1,4 @@
 class MiqSmartProxyWorker::Runner < MiqQueueWorkerBase::Runner
-  def do_before_work_loop
-    @tid = start_heartbeat_thread
-  end
-
   def before_exit(message, _exit_code)
     @exit_requested = true
     #
@@ -54,18 +50,24 @@ def heartbeat_thread
     end
   end
 
-  def do_work
-    if @tid.nil? || !@tid.alive?
-      unless @tid.try(:status)
-        dead_tid, @tid = @tid, nil
-        _log.info("#{log_prefix} Waiting for the Heartbeat Thread to exit...")
-        dead_tid.join # raise the exception the dead thread failed with
-      end
+  def ensure_heartbeat_thread_started
+    # start the thread and return if it has not been started yet
+    if @tid.nil?
+      @tid = start_heartbeat_thread
+      return
+    end
+
+    # if the thread is dead, start a new one and kill it if it is aborting
+    if !@tid.alive?
+      @tid.kill if @tid.status == "aborting"
 
       _log.info("#{log_prefix} Heartbeat Thread gone. Restarting...")
       @tid = start_heartbeat_thread
     end
+  end
 
+  def do_work
+    ensure_heartbeat_thread_started
     super
   end
 end

spec/models/miq_smart_proxy_worker/runner_spec.rb

@@ -0,0 +1,50 @@
+describe MiqSmartProxyWorker::Runner do
+  subject { described_class.allocate }
+
+  describe "#ensure_heartbeat_thread_started" do
+    let(:thread) { double(Thread) }
+
+    it "starts the heartbeat thread if it has never been started" do
+      expect(subject).to receive(:start_heartbeat_thread).and_return(thread)
+      subject.ensure_heartbeat_thread_started
+      expect(subject.instance_variable_get(:@tid)).to eq(thread)
+    end
+
+    it "restarts the heartbeat thread if it has exited" do
+      subject.instance_variable_set(:@tid, thread)
+      expect(thread).to receive(:alive?).and_return(false)
+      expect(thread).to receive(:status).and_return(false)
+
+      new_thread = double(Thread)
+      expect(subject).to receive(:start_heartbeat_thread).and_return(new_thread)
+
+      subject.ensure_heartbeat_thread_started
+      expect(subject.instance_variable_get(:@tid)).to eq(new_thread)
+    end
+
+    it "restarts the heartbeat thread if it has failed" do
+      subject.instance_variable_set(:@tid, thread)
+      expect(thread).to receive(:alive?).and_return(false)
+      expect(thread).to receive(:status).and_return(nil)
+
+      new_thread = double(Thread)
+      expect(subject).to receive(:start_heartbeat_thread).and_return(new_thread)
+
+      subject.ensure_heartbeat_thread_started
+      expect(subject.instance_variable_get(:@tid)).to eq(new_thread)
+    end
+
+    it "kills the heartbeat thread if it is aborting and restarts it" do
+      subject.instance_variable_set(:@tid, thread)
+      expect(thread).to receive(:alive?).and_return(false)
+      expect(thread).to receive(:status).and_return("aborting")
+      expect(thread).to receive(:kill)
+
+      new_thread = double(Thread)
+      expect(subject).to receive(:start_heartbeat_thread).and_return(new_thread)
+
+      subject.ensure_heartbeat_thread_started
+      expect(subject.instance_variable_get(:@tid)).to eq(new_thread)
+    end
+  end
+end