If we can't update_attributes on a queue row, set state to error

https://bugzilla.redhat.com/show_bug.cgi?id=1429747

In the reported bug, we had a Rails 4.2 era class [1] serialized in the args
column of a miq_queue row.  This class was removed in rails 5.0.0 [2],
so we'd be unable to deserialize the column with:

ArgumentError: undefined class/module ActiveRecord::ConnectionAdapters::PostgreSQL::OID::Integer

If we're unable to update_attributes because a column can't be
deserialized, the message can't be handled by a worker, the worker dies,
and the message remains in the miq_queue for another worker to try and
also fail on.

Instead, if update_attributes fails, we can try to set just the state
column to 'error'.  In this way, the server will not try to dispatch the
same queue multiple times.  We clear errored messages at server boot, so
we can clean them up then.

[1] ActiveRecord::ConnectionAdapters::PostgreSQL::OID::Integer
[2] rails/rails@aafee23

app/models/miq_queue_worker_base/runner.rb

@@ -35,7 +35,7 @@ def thresholds_exceeded?
   def get_message_via_drb
     loop do
       begin
-        msg_id, lock_version = @worker_monitor_drb.get_queue_message(@worker.pid)
+        msg_id, lock_version = worker_monitor_drb.get_queue_message(@worker.pid)
       rescue DRb::DRbError => err
         do_exit("Error communicating with WorkerMonitor because <#{err.message}>", 1)
       end
@@ -64,6 +64,7 @@ def get_message_via_drb
         _log.debug("#{log_prefix} #{MiqQueue.format_short_log_msg(msg)} stale, retrying...")
         next
       rescue => err
+        msg.update_column(:state, MiqQueue::STATUS_ERROR)
         raise _("%{log} \"%{error}\" attempting to get next message") % {:log => log_prefix, :error => err}
       end
     end

spec/factories/miq_worker.rb

@@ -4,6 +4,7 @@
     status "ready"
   end
 
+  factory :miq_generic_worker, :class => "MiqGenericWorker", :parent => :miq_worker
   factory :miq_ui_worker, :class => "MiqUiWorker", :parent => :miq_worker
   factory :miq_websocket_worker, :class => "MiqWebsocketWorker", :parent => :miq_worker
 

spec/models/miq_queue_worker_base/runner_spec.rb

@@ -0,0 +1,24 @@
+describe MiqQueueWorkerBase::Runner do
+  context "#get_message_via_drb" do
+    let(:server) { EvmSpecHelper.local_miq_server }
+    let(:worker) { FactoryGirl.create(:miq_generic_worker, :miq_server => server, :pid => 123) }
+    let(:runner) do
+      allow_any_instance_of(MiqQueueWorkerBase::Runner).to receive(:sync_active_roles)
+      allow_any_instance_of(MiqQueueWorkerBase::Runner).to receive(:sync_config)
+      allow_any_instance_of(MiqQueueWorkerBase::Runner).to receive(:set_connection_pool_size)
+      described_class.new(:guid => worker.guid)
+    end
+
+    it "sets message to 'error' and raises for unhandled exceptions" do
+      # simulate what may happen if invalid yaml is deserialized
+      allow_any_instance_of(MiqQueue).to receive(:args).and_raise(ArgumentError)
+      q1 = FactoryGirl.create(:miq_queue)
+      allow(runner)
+        .to receive(:worker_monitor_drb)
+        .and_return(double(:get_queue_message => [q1.id, q1.lock_version]))
+
+      expect { runner.get_message_via_drb }.to raise_error(StandardError)
+      expect(q1.reload.state).to eql(MiqQueue::STATUS_ERROR)
+    end
+  end
+end